Hi Sophie,
Thanks for your feedback.
Completing the Damien's comments here for points S1 and S5B.
S1:
I'm confused -- are you saying that we're introducing a new kind of
ProducerRecord class for this?
I am wondering if it makes sense to alter the ProducerRecord from Clients API with a
"deadLetterQueueTopicName" attribute dedicated to Kafka Streams DLQ.
Adding "deadLetterQueueTopicName" as an additional parameter to
"withDeadLetterQueueRecord" is a good option, and may allow users to send records to
different DLQ topics depending on conditions:
@Override
public ProductionExceptionHandlerResponse handle(final ProcessingContext
context,
ProducerRecord<byte[], byte[]> record,
Exception exception) {
if (condition1) {
return ProductionExceptionHandlerResponse.CONTINUE
.withDeadLetterQueueRecord(record, "dlq-topic-a");
}
if (condition2) {
return ProductionExceptionHandlerResponse.CONTINUE
.withDeadLetterQueueRecord(record, "dlq-topic-b");
}
return ProductionExceptionHandlerResponse.CONTINUE
.withDeadLetterQueueRecord(record, "dlq-topic-c");
}
S5B:
I was having a bit of trouble understanding what the behavior would be if someone
configured a "errors.deadletterqueue.topic.name" but didn't implement the
handlers.
The provided LogAndContinueExceptionHandler, LogAndFailExceptionHandler and
DefaultProductionExceptionHandler should be able to tell if records should be
sent to DLQ or not.
The "errors.deadletterqueue.topic.name" takes place to:
* Specifying if the provided handlers should or should not send records to DLQ.
* If the value is empty, the handlers should not send records to DLQ.
* If the value is not empty, the handlers should send records to DLQ.
* Define the name of the DLQ topic that should be used by the provided handlers.
Thus, if "errors.deadletterqueue.topic.name" is defined, the provided handlers
should return either:
* CONTINUE.withDeadLetterQueueRecord(record, defaultDeadLetterQueue)
* FAIL.withDeadLetterQueueRecord(record, defaultDeadLetterQueue).
If "errors.deadletterqueue.topic.name" is defined but neither
DeserializationExceptionHandler nor ProductionExceptionHandler classes are defined in the
configuration, then nothing should happen as sending to DLQ is based on handlers’
response.
When providing custom handlers, users would have the possibility to return:
* FAIL
* CONTINUE
* FAIL.withDeadLetterQueueRecord(record, "dlq-topic")
* CONTINUE.withDeadLetterQueueRecord(record, "dlq-topic")
A DLQ topic name is currently required using the two last response types.
I am wondering if it could benefit users to ease the use of the default DLQ topic
"errors.deadletterqueue.topic.name" when implementing custom handlers, with
such kind of implementation:
* FAIL.withDefaultDeadLetterQueueRecord(record)
* CONTINUE.withDefaultDeadLetterQueueRecord(record)
Regards,
Loïc
De : Damien Gasparina <d.gaspar...@gmail.com>
Envoyé : dimanche 14 avril 2024 20:24
À : dev@kafka.apache.org
Objet : [EXT] Re: [DISCUSS] KIP-1034: Dead letter queue in Kafka Streams
Warning External sender Do not click on any links or open any attachments
unless you trust the sender and know the content is safe.
Hi Sophie,
Thanks a lot for your feedback and your detailed comments.
S1.
I'm confused -- are you saying that we're introducing a new kind of
ProducerRecord class for this?
Sorry for the poor wording, that's not what I meant. While writing the
KIP, I was hesitating between 1. leveraging the Kafka Producer
ProducerRecord, 2. the Kafka Streams ProducerRecord + a topic name in
a separate parameter, 3. a new custom interface (e.g.
DeadLetterQueueRecord).
As the KafkaProducer ProducerRecord is not used in the Kafka Streams
API (except ProductionExceptionHandler) and I would like to avoid a
new interface if not strictly required, I leaned toward option 2.
Thinking about it, maybe option 1. would be best, but I assume it
could create confusion with KafkaStreams ProducerRecord. Let me sleep
on it.
S2. I agree. Following the discussion in KIP-1033 and KIP-1034 and
your point in S4, it seems more and more likely that we will create a
new container class containing only the metadata for the exception
handlers. To be consistent, I think we should use this new
implementation in all exception handlers.
The only issue I could think off is that the new interface would
expose less data than the current ProcessorContext in the
DeserializationException(e.g. stateDir(), metrics(), getStateStore()),
thus it could be hard for some users to migrate to the new interface.
I do expect that only a few users would be impacted as the javadoc is
very clear: `Note, that the passed in {@link ProcessorContext} only
allows access to metadata like the task ID.`
S3. I completely agree with you, it is something that might not be
trivial and should be thoroughly covered by unit tests during the
implementation.
S4. Good point, I did not notice that the ProductionExceptionHandler
is also invoked in the producer.send() callback.
Capturing the ProcessingContext for each in-flight message is probably
not possible. I think there is no other way to write a custom
container class holding only the metadata that are essentials, I am
thinking of storing the following attributes: source topic, partition,
offset, rawKey, rawValue and taskId.
Those metadata should be relatively small, but I assume that there
could be a high number of in-flight messages, especially with at least
once processing guarantee. Do you think it would be fine memory wise?
S5. As many exceptions are only accessible in exception handlers, and
we wanted to 1) allow users to customize the DLQ records and 2) have a
suitable DLQ out of the box implementation, we felt it natural to rely
on exception handlers, that's also why we created KIP-1033.
Piggybacking on the enum response was the cleanest way we could think
off, but we are completely open to suggestions.
S5a. Completely agree with you on this point, for this DLQ approach to
be complete, the ProcessingExceptionHandler introduced in KIP-1033 is
required. KIP-1033 is definitely our first priority. We decided to
kick-off the KIP-1034 discussion as we expected the discussions to be
dynamic and could potentially impact some choices of KIP-1033.
S5b. In this KIP, we wanted to 1. provide as much flexibility to the
user as possible; 2. provide a good default implementation
for the DLQ without having to write custom exception handlers.
For the default implementation, we introduced a new configuration:
errors.deadletterqueue.topic.name.
If this configuration is set, it changes the behavior of the provided
exception handlers to return a DLQ record containing the raw key/value
+ headers + exception metadata in headers.
If the out of the box implementation is not suitable for a user, e.g.
the payload needs to be masked in the DLQ, it could implement their
own exception handlers. The errors.deadletterqueue.topic.name would
only impact Kafka Streams bundled exception handlers (e.g.
org.apache.kafka.streams.errors;.LogAndContinueExceptionHandler)
Let me update the KIP to make it clear and also provide examples.
S6/S7. Good point, mea culpa for the camel case, it must have been a
sugar rush :)
Thanks again for your detailed comments and pointing out S4
(production exception & Processing Context)!
Cheers,
Damien
This email was screened for spam and malicious content but exercise caution
anyway.
On Fri, 12 Apr 2024 at 22:04, Sophie Blee-Goldman
<sop...@responsive.dev<mailto:sop...@responsive.dev>> wrote:
Thanks for the KIP, this will make a lot of people very happy.
Wanted to chime in on a few points that have been raised so far and add
some of my own (numbering with an S to distinguish my points from the
previous ones)
S1.
1.a I really meant ProducerRecord, that's the class used to forward to
downstream processors in the PAPI. The only information missing in
this class is the topic name. I also considered relying on the Kafka
Producer ProducerRecord, but I assume it would not be consistent with
the KafkaStreams API.
I'm confused -- are you saying that we're introducing a new kind of
ProducerRecord class for this? Why not just use the existing one, ie the
o.a.k.clients.producer.ProducerRecord class? This is what the
ProductionExceptionHandler uses, so it's definitely "consistent". In other
words, we can remove the "String deadLetterQueueTopicName"
S2.
I think this would be a good opportunity to also deprecate the existing
#handle method of the DeserializationExceptionHandler, and replace it with
one that uses a ProcessingContext instead of the ProcessorContext. Partly
for the same reasons about guarding access to the #forward methods, partly
because this method needs to be migrated to the new PAPI interface
anyways, and ProcessingContext is part of the new one.
S3.
Regarding 2a. -- I'm inclined to agree that records which a Punctuator
failed to produce should also be sent to the DLQ via the
ProductionExceptionHandler. Users will just need to be careful about
accessing certain fields of the ProcessingContext that aren't available in
the punctuator, and need to check the Optional returned by the
ProcessingContext#recordMetadata API.
Also, from an implementation standpoint, it will be really hard to
distinguish between a record created by a punctuator vs a processor from
within the RecordCollector, which is the class that actually handles
sending records to the Streams Producer and invoking the
ProductionExceptionHandler. This is because the RecordCollector is at the
"end" of the topology graph and doesn't have any context about which of the
upstream processors actually attempted to forward a record.
This in itself is at least theoretically solvable, but it leads into my
first major new point:
S4:
I'm deeply worried about passing the ProcessingContext in as a means of
forwarding metadata. The problem is that the processing/processor context
is a mutable class and is inherently meaningless outside the context of a
specific task. And when I said earlier that the RecordCollector sits at
the "end" of the topology, I meant that it's literally outside the task's
subtopology and is used/shared by all tasks on that StreamThread. So to
begin with, there's no guarantee what will actually be returned for
essential methods such as the new #rawSourceKey/Value or the existing
#recordMetadata
For serialization exceptions it'll probably be correct, but for general
send errors it almost definitely won't be. In short, this is because we
send records to the producer after the sink node, but don't check for send
errors right away since obviously it takes some time for the producer to
actually send. In other words, sending/producing records is actually done
asynchronously with processing, and we simply check for errors on any
previously-sent records
during the send on a new record in a sink node. This means the context we
would be passing in to a (non-serialization) exception would pretty much
always correspond not the the record that experienced the error, but the
random record that happened to be being sent when we checked and saw the
error for the failed record.
This discrepancy, in addition to the whole "sourceRawKey/Value and
recordMetadata are null for punctuators" issue, seems like an
insurmountable inconsistency that is more likely to cause users confusion
or problems than be helpful.
We could create a new metadata object and copy over the relevant info from
the ProcessingContext, but I worry that has the potential to explode memory
since we'd need to hold on to it for all in-flight records up until they
are either successfully sent or failed and passed in to the
ProductionExceptionHandler. But if the metadata is relatively small, it's
probably fine. Especially if it's just the raw source key/value. Are
there any other parts of the ProcessingContext you think should be made
available?
Note that this only applies to the ProductionExceptionHandler, as the
DeserializationExceptionHandler (and the newly proposed
ProcessingExceptionHandler) would both be invoked immediately and therefore
with the failed record's context. However, I'm also a bit uncomfortable
with adding the rawSourceKey/rawSourceValue to the ProcessingContext. So
I'd propose to just wrap those (and any other metadata you might want) in a
container class and pass that in instead of the ProcessingContext, to all
of the exception handlers.
S5:
For some reason I'm finding the proposed API a little bit awkward, although
it's entirely possible that the problem is with me, not the proposal :)
Specifically I'm struggling with the approach of piggybacking on the
exception handlers and their response enums to dictate how records are
forwarded to the DLQ. I think this comes down to two things, though again,
these aren't necessarily problems with the API and probably just need to be
hashed out:
S5a.
When I envision a DLQ, to me, the most common use case would be to forward
input records that failed somewhere along the processing graph. But it
seems like all the focus here is on the two far ends of the subtopology --
the input/consumer, and the output/producer. I get that
the ProcessingExceptionHandler is really the missing piece here, and it's
hard to say anything specific since it's not yet accepted, but maybe a
somewhat more concrete example would help. FWIW I think/hope to get that
KIP accepted and implementation ASAP, so I'm not worried about the "what if
it doesn't happen" case -- more just want to know what it will look like
when it does. Imo it's fine to build KIPs on top of future ones, it feels
clear that this part will just have to wait for that KIP to actually be
added.
S5b:
Why do users have to define the entire ProducerRecord -- shouldn't Streams
handle all this for them? Or will we just automatically send every record
on failure to the default global DLQ, and users only have to implement the
handlers if they want to change the headers or send to a different topic? I
was having a bit of trouble understanding what the behavior would be if
someone configured a "errors.deadletterqueue.topic.name" but didn't
implement the handlers. Apologies if it's somewhere in the KIP and I
happened to miss it!
Either way, I really think an example would help me to better imagine what
this will look like in practice, and evaluate whether it actually involves
as much overhead as I'm worried it will. Can you add a section that
includes a basic implementation of all the features here? Nothing too
complicated, just the most bare-bones code needed to actually implement
forwarding to a dead-letter-queue via the handlers.
Lastly, two super small things:
S6:
We use camel case in Streams, so it should be rawSourceKey/Value rather
than raw_source_key/value
S7:
Can you add javadocs for the #withDeadLetterQueueRecord? For example, it
seems to me that if the topic to be sent to here is different than the
default/global DLQ, then the user will need to make sure to have created
this themselves up front.
That's it from me...sorry for the long response, it's just because I'm
excited for this feature and have been waiting on a KIP for this for years.
Cheers,
Sophie
On Fri, Apr 12, 2024 at 11:10 AM Damien Gasparina
<d.gaspar...@gmail.com<mailto:d.gaspar...@gmail.com>>
wrote:
Hi Andrew,
Thanks a lot for your review, plenty of good points!
11. Typo fixed, good cach.
12. I do agree with you and Nick also mentioned it, I updated the KIP
to mention that context headers should be forwarded.
13. Good catch, to be consistent with KIP-298, and without a strong
opinion from my side, I updated the KIP with your prefix proposal.
14. I am not sure about this point, a big difference between KIP-298
and this one is that the handlers can easily be overridden, something
that is not doable in Kafka Connect.
If someone would like a different behavior, e.g. to mask the payload
or include further headers, I think we should encourage them to write
their own exception handlers to build the DLQ Record the way they
expect.
15. Yeah, that's a good point, I was not fully convinced about putting
a String in it, I do assume that "null" is also a valid value. I do
assume that the Stacktrace and the Exception in this case are the key
metadata for the user to troubleshoot the problem.
I updated the KIP to mention that the value should be null if
triggered in a punctuate.
16. I added a session to mention that Kafka Streams would not try to
automatically create the topic and the topic should either be
automatically created, or pre-created.
17. If a DLQ record can not be sent, the exception should go to the
uncaughtExceptionHandler. Let me clearly state it in the KIP.
On Fri, 12 Apr 2024 at 17:25, Damien Gasparina
<d.gaspar...@gmail.com<mailto:d.gaspar...@gmail.com>>
wrote:
Hi Nick,
1. Good point, that's less impactful than a custom interface, I just
updated the KIP with the new signature.
1.a I really meant ProducerRecord, that's the class used to forward to
downstream processors in the PAPI. The only information missing in
this class is the topic name. I also considered relying on the Kafka
Producer ProducerRecord, but I assume it would not be consistent with
the KafkaStreams API.
2. Agreed
2.a I do think exceptions occurring during punctuate should be
included in the DLQ.
Even if building a suitable payload is almost impossible, even with
custom code; those exceptions are still fatal for Kafka Streams by
default and are something that can not be ignored safely.
I do assume that most users would want to be informed if an error
happened during a punctuate, even if only the metadata (e.g.
stacktrace, exception) is provided.
I am only concerned flooding the DLQ topic as, if a scheduled
operation failed, very likely it will fails during the next
invocation, but
4. Good point, I clarified the wording in the KIP to make it explicit.
5. Good point, I will clearly mention that it is out of scope as part
of the KIP and might not be as trivial as people could expect. I will
update the KIP once I do have some spare time.
6. Oh yeah, I didn't think about it, but forwarding input headers
would definitely make sense. Confluent Schema Registry ID is actually
part of the payload, but many correlation ID and technical metadata
are passed through headers, it makes sense to forward them, specially
as it is the default behavior of Kafka Streams,
On Fri, 12 Apr 2024 at 15:25, Nick Telford
<nick.telf...@gmail.com<mailto:nick.telf...@gmail.com>>
wrote:
Hi Damien and Sebastien,
1.
I think you can just add a `String topic` argument to the existing
`withDeadLetterQueueRecord(ProducerRecord<byte[], byte[]>
deadLetterQueueRecord)` method, and then the implementation of the
exception handler could choose the topic to send records to using
whatever
logic the user desires. You could perhaps provide a built-in
implementation
that leverages your new config to send all records to an untyped DLQ
topic?
1a.
BTW you have a typo: in your DeserializationExceptionHandler, the type
of
your `deadLetterQueueRecord` argument is `ProducerRecord`, when it
should
probably be `ConsumerRecord`.
2.
Agreed. I think it's a good idea to provide an implementation that
sends to
a single DLQ by default, but it's important to enable users to
customize
this with their own exception handlers.
2a.
I'm not convinced that "errors" (e.g. failed punctuate) should be sent
to a
DLQ topic like it's a bad record. To me, a DLQ should only contain
records
that failed to process. I'm not even sure how a user would
re-process/action one of these other errors; it seems like the purview
of
error logging to me?
4.
My point here was that I think it would be useful for the KIP to
contain an
explanation of the behavior both with KIP-1033 and without it. i.e.
clarify
if/how records that throw an exception in a processor are handled. At
the
moment, I'm assuming that without KIP-1033, processing exceptions
would not
cause records to be sent to the DLQ, but with KIP-1033, they would. If
this
assumption is correct, I think it should be made explicit in the KIP.
5.
Understood. You may want to make this explicit in the documentation for
users, so they understand the consequences of re-processing data sent
to
their DLQ. The main reason I raised this point is it's something that's
tripped me up in numerous KIPs that that committers frequently remind
me
of; so I wanted to get ahead of it for once! :D
And one new point:
6.
The DLQ record schema appears to discard all custom headers set on the
source record. Is there a way these can be included? In particular, I'm
concerned with "schema pointer" headers (like those set by Schema
Registry), that may need to be propagated, especially if the records
are
fed back into the source topics for re-processing by the user.
Regards,
Nick
On Fri, 12 Apr 2024 at 13:20, Damien Gasparina
<d.gaspar...@gmail.com<mailto:d.gaspar...@gmail.com>>
wrote:
Hi Nick,
Thanks a lot for your review and your useful comments!
1. It is a good point, as you mentioned, I think it would make sense
in some use cases to have potentially multiple DLQ topics, so we
should provide an API to let users do it.
Thinking out-loud here, maybe it is a better approach to create a new
Record class containing the topic name, e.g. DeadLetterQueueRecord
and
changing the signature to
withDeadLetterQueueRecords(Iteratable<DeadLetterQueueRecord>
deadLetterQueueRecords) instead of
withDeadLetterQueueRecord(ProducerRecord<byte[], byte[]>
deadLetterQueueRecord). What do you think? DeadLetterQueueRecord
would
be something like "class DeadLetterQueueRecord extends
org.apache.kafka.streams.processor.api;.ProducerRecords { String
topic; /* + getter/setter + */ } "
2. I think the root question here is: should we have one DLQ topic or
multiple DLQ topics by default. This question highly depends on the
context, but implementing a default implementation to handle multiple
DLQ topics would be opinionated, e.g. how to manage errors in a
punctuate?
I think it makes sense to have the default implementation writing all
faulty records to a single DLQ, that's at least the approach I used
in
past applications: one DLQ per Kafka Streams application. Of course
the message format could change in the DLQ e.g. due to the source
topic, but those DLQ records will be very likely troubleshooted, and
maybe replay, manually anyway.
If a user needs to have multiple DLQ topics or want to enforce a
specific schema, it's still possible, but they would need to
implement
custom Exception Handlers.
Coming back to 1. I do agree that it would make sense to have the
user
set the DLQ topic name in the handlers for more flexibility.
3. Good point, sorry it was a typo, the ProcessingContext makes much
more sense here indeed.
4. I do assume that we could implement KIP-1033 (Processing exception
handler) independently from KIP-1034. I do hope that KIP-1033 would
be
adopted and implemented before KIP-1034, but if that's not the case,
we could implement KIP-1034 indepantly and update KIP-1033 to include
the DLQ record afterward (in the same KIP or in a new one if not
possible).
5. I think we should be clear that this KIP only covers the DLQ
record
produced.
Everything related to replay messages or recovery plan should be
considered out-of-scope as it is use-case and error specific.
Let me know if that's not clear, there are definitely points that
highly debatable.
Cheers,
Damien
On Fri, 12 Apr 2024 at 13:00, Nick Telford
<nick.telf...@gmail.com<mailto:nick.telf...@gmail.com>>
wrote:
Oh, and one more thing:
5.
Whenever you take a record out of the stream, and then potentially
re-introduce it at a later date, you introduce the potential for
record
ordering issues. For example, that record could have been destined
for a
Window that has been closed by the time it's re-processed. I'd
like to
see
a section that considers these consequences, and perhaps make
those risks
clear to users. For the record, this is exactly what sunk KIP-990,
which
was an alternative approach to error handling that introduced the
same
issues.
Cheers,
Nick
On Fri, 12 Apr 2024 at 11:54, Nick Telford <nick.telf...@gmail.com
<mailto:nick.telf...@gmail.com%0b>> > >
wrote:
Hi Damien,
Thanks for the KIP! Dead-letter queues are something that I
think a
lot of
users would like.
I think there are a few points with this KIP that concern me:
1.
It looks like you can only define a single, global DLQ for the
entire
Kafka Streams application? What about applications that would
like to
define different DLQs for different data flows? This is
especially
important when dealing with multiple source topics that have
different
record schemas.
2.
Your DLQ payload value can either be the record value that
failed, or
an
error string (such as "error during punctuate"). This is likely
to
cause
problems when users try to process the records from the DLQ, as
they
can't
guarantee the format of every record value will be the same.
This is
very
loosely related to point 1. above.
3.
You provide a ProcessorContext to both exception handlers, but
state
they
cannot be used to forward records. In that case, I believe you
should
use
ProcessingContext instead, which statically guarantees that it
can't be
used to forward records.
4.
You mention the KIP-1033 ProcessingExceptionHandler, but what's
the
plan
if KIP-1033 is not adopted, or if KIP-1034 lands before 1033?
Regards,
Nick
On Fri, 12 Apr 2024 at 11:38, Damien Gasparina <
d.gaspar...@gmail.com<mailto:d.gaspar...@gmail.com>>
wrote:
In a general way, if the user does not configure the right ACL,
that
would be a security issue, but that's true for any topic.
This KIP allows users to configure a Dead Letter Queue without
writing
custom Java code in Kafka Streams, not at the topic level.
A lot of applications are already implementing this pattern,
but the
required code to do it is quite painful and error prone, for
example
most apps I have seen created a new KafkaProducer to send
records to
their DLQ.
As it would be disabled by default for backward compatibility,
I doubt
it would generate any security concern.
If a user explicitly configures a Deal Letter Queue, it would
be up to
him to configure the relevant ACLs to ensure that the right
principal
can access it.
It is already the case for all internal, input and output Kafka
Streams topics (e.g. repartition, changelog topics) that also
could
contain confidential data, so I do not think we should
implement a
different behavior for this one.
In this KIP, we configured the default DLQ record to have the
initial
record key/value as we assume that it is the expected and wanted
behavior for most applications.
If a user does not want to have the key/value in the DLQ record
for
any reason, they could still implement exception handlers to
build
their own DLQ record.
Regarding ACL, maybe something smarter could be done in Kafka
Streams,
but this is out of scope for this KIP.
On Fri, 12 Apr 2024 at 11:58, Claude Warren
<cla...@xenei.com<mailto:cla...@xenei.com>>
wrote:
My concern is that someone would create a dead letter queue
on a
sensitive
topic and not get the ACL correct from the start. Thus
causing
potential
confidential data leak. Is there anything in the proposal
that
would
prevent that from happening? If so I did not recognize it as
such.
On Fri, Apr 12, 2024 at 9:45 AM Damien Gasparina <
d.gaspar...@gmail.com<mailto:d.gaspar...@gmail.com>
wrote:
Hi Claude,
In this KIP, the Dead Letter Queue is materialized by a
standard
and
independant topic, thus normal ACL applies to it like any
other
topic.
This should not introduce any security issues, obviously,
the
right
ACL would need to be provided to write to the DLQ if
configured.
Cheers,
Damien
On Fri, 12 Apr 2024 at 08:59, Claude Warren, Jr
<claude.war...@aiven.io.invalid<mailto:claude.war...@aiven.io.invalid>> wrote:
I am new to the Kafka codebase so please excuse any
ignorance
on my
part.
When a dead letter queue is established is there a
process to
ensure that
it at least is defined with the same ACL as the original
queue?
Without
such a guarantee at the start it seems that managing dead
letter
queues
will be fraught with security issues.
On Wed, Apr 10, 2024 at 10:34 AM Damien Gasparina <
d.gaspar...@gmail.com<mailto:d.gaspar...@gmail.com>
wrote:
Hi everyone,
To continue on our effort to improve Kafka Streams error
handling, we
propose a new KIP to add out of the box support for Dead
Letter
Queue.
The goal of this KIP is to provide a default
implementation
that
should be suitable for most applications and allow
users to
override
it if they have specific requirements.
In order to build a suitable payload, some additional
changes
are
included in this KIP:
1. extend the ProcessingContext to hold, when
available, the
source
node raw key/value byte[]
2. expose the ProcessingContext to the
ProductionExceptionHandler,
it is currently not available in the handle parameters.
Regarding point 2., to expose the ProcessingContext to
the
ProductionExceptionHandler, we considered two choices:
1. exposing the ProcessingContext as a parameter in
the
handle()
method. That's the cleanest way IMHO, but we would need
to
deprecate
the old method.
2. exposing the ProcessingContext as an attribute in
the
interface.
This way, no method is deprecated, but we would not be
consistent
with
the other ExceptionHandler.
In the KIP, we chose the 1. solution (new handle
signature
with
old
one deprecated), but we could use other opinions on
this part.
More information is available directly on the KIP.
KIP link:
https://cwiki.apache.org/confluence/display/KAFKA/KIP-1034%3A+Dead+letter+queue+in+Kafka+Streams<https://cwiki.apache.org/confluence/display/KAFKA/KIP-1034%3A+Dead+letter+queue+in+Kafka+Streams><https://cwiki.apache.org/confluence/display/KAFKA/KIP-1034%3A+Dead+letter+queue+in+Kafka+Streams<https://cwiki.apache.org/confluence/display/KAFKA/KIP-1034%3A+Dead+letter+queue+in+Kafka+Streams>>
Feedbacks and suggestions are welcome,
Cheers,
Damien, Sebastien and Loic
--
LinkedIn:
http://www.linkedin.com/in/claudewarren<http://www.linkedin.com/in/claudewarren><http://www.linkedin.com/in/claudewarren<http://www.linkedin.com/in/claudewarren>>
