Hi Bruno,
We focused our effort (well, mostly Seb and Loic :)) on KIP-1033,
that's why not much progress has been made on this one yet.
Regarding your points:
B1: It is up to the user to specify the DLQ topic name and to
implement a potential differentiation. I tend to think that having one
DLQ per application ID is the wisest, but I encountered cases and
applications that preferred having a shared DLQ topic between multiple
applications, e.g. to reduce the number of partitions, or to ease
monitoring
B2 : Goot catch, it should be
ERRORS_DEADLETTERQUEUE_TOPIC_NAME_CONFIG, we removed the "DEFAULT"
prefix during the discussion, looks like I forgot to update all
occurrences in the KIP.
B3 :The trigger for sending to the DLQ would be if
ERRORS_DEADLETTERQUEUE_TOPIC_NAME_CONFIG is set OR if the user
implemented a custom exception handler that returns DLQ records.
ERRORS_DEADLETTERQUEUE_TOPIC_NAME_CONFIG would just override the
behavior of the default handler, thus custom exception handlers will
completely ignore this parameter.
I think it's a good trade-off between providing a production-ready
default implementation, yet providing sufficient flexibility for
complex use-cases.
This behavior definitely needs to be documented, but I guess it's safe
to push the responsibility of the DLQ records to the user if they
implement custom handlers.
Cheers,
Damien
On Tue, 11 Jun 2024 at 17:00, Bruno Cadonna <cado...@apache.org> wrote:
>
> Hi,
>
> since there was not too much activity in this thread recently, I was
> wondering what the status of this discussion is.
>
> I cannot find the examples in the KIP Sébastien mentioned in the last
> message to this thread. I can also not find the corresponding definition
> of the following method call in the KIP:
>
> FAIL.withDeadLetterQueueRecord(record, "dlq-topic")
>
> I have also some comments:
>
> B1
> Did you consider to prefix the dead letter queue topic names with the
> application ID to distinguish the topics between Streams apps? Or is the
> user responsible for the differentiation? If the user is responsible, we
> risk that faulty records of different Streams apps end up in the same
> dead letter queue.
>
> B2
> Is the name of the dead letter queue topic config
> DEFAULT_ERRORS_DEADLETTERQUEUE_TOPIC_NAME_CONFIG or
> ERRORS_DEADLETTERQUEUE_TOPIC_NAME_CONFIG? In the KIP both names are used.
>
> B3
> What is exactly the trigger to send a record to the dead letter queue?
> Is setting ERRORS_DEADLETTERQUEUE_TOPIC_NAME_CONFIG or is it adding a
> record to the return value of the exception handler?
> What happens if I set ERRORS_DEADLETTERQUEUE_TOPIC_NAME_CONFIG but do
> not add a record to the return value of the handler? What happens if I
> do not set ERRORS_DEADLETTERQUEUE_TOPIC_NAME_CONFIG but add a record to
> the return value of the handler?
>
> Best,
> Bruno
>
> On 4/22/24 10:19 PM, Sebastien Viale wrote:
> > Hi,
> >
> > Thanks for your remarks
> >
> > L1. I would say "who can do the most can do the least", even though most
> > people will fail and stop, we found it interesting to offer the possibility
> > to fail-and-send-to-DLQ
> >
> > L2: We did not consider extending the TimestampExtractor because we
> > estimate it out of scope for this KIP. Perhaps it will be possible to
> > include it in an ExceptionHandler later.
> >
> > L3: we will include an example in the KIP, but as we mentioned earlier, the
> > DLQ topic can be different in each custom Exception Handler:
> >
> > When providing custom handlers, users would have the possibility to return:
> > * FAIL
> > * CONTINUE
> > * FAIL.withDeadLetterQueueRecord(record, "dlq-topic")
> > * CONTINUE.withDeadLetterQueueRecord(record, "dlq-topic")
> >
> > cheers !
> > Sébastien
> >
> >
> > ________________________________
> > De : Lucas Brutschy <lbruts...@confluent.io.INVALID>
> > Envoyé : lundi 22 avril 2024 14:36
> > À : dev@kafka.apache.org <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!
> >
> > Thanks for the KIP, great stuff.
> >
> > L1. I was a bit confused that the default configuration (once you set
> > a DLQ topic) is going to be fail-and-send-to-DLQ, if I understood
> > correctly. Is this something that will be a common use-case, and is it
> > a configuration that we want to encourage? It expected that you either
> > want to fail or skip-and-send-to-DLQ.
> >
> > L2. Have you considered extending the `TimestampExtractor` interface
> > so that it can also produce to DLQ? AFAIK it's not covered by any of
> > the existing exception handlers, but it can cause similar failures
> > (potentially custom logic, depends on validity input record). There
> > could also be a default implementation as a subclass of
> > `ExtractRecordMetadataTimestamp`.
> >
> > L3. It would be nice to include an example of how to produce to
> > multiple topics in the KIP, as I can imagine that this will be a
> > common use-case. I wasn't sure how much code would be involved to make
> > it work. If a lot of code is required, we may want to consider
> > exposing some utils that make it easier.
> >
> > Cheers,
> > Lucas
> >
> > This email was screened for spam and malicious content but exercise caution
> > anyway.
> >
> >
> >
> > On Sun, Apr 21, 2024 at 7:58 PM Damien Gasparina <d.gaspar...@gmail.com>
> > wrote:
> >>
> >> Hi everyone,
> >>
> >> Following all the discussion on this KIP and KIP-1033, we introduced a
> >> new container class containing only processing context metadata:
> >> ProcessingMetadata. This new container class is actually part of
> >> KIP-1033, thus, I added a hard dependency for this KIP on KIP-1033, I
> >> think it's the wisest implementation wise.
> >>
> >> I also clarified the interface of the enums:
> >> withDeadLetterQueueRecords(Iterable<org.apache.kafka.clients.producer.ProducerRecord<byte[],
> >> byte[]>> deadLetterQueueRecords) . Very likely most users would just
> >> send one DLQ record, but there might be specific use-cases and what
> >> can do more can do less, so I added an Iterable.
> >>
> >> I took some time to think about the impact of storing the
> >> ProcessingMetadata on the ProductionExceptionHandler. I think storing
> >> the topic/offset/partition should be fine, but I am concerned about
> >> storing the rawSourceKey/Value. I think it could impact some specific
> >> use-cases, for example, a high-throughput Kafka Streams application
> >> "counting" messages could have huge source input messages, and very
> >> small sink messages, here, I assume storing the rawSourceKey/Value
> >> could significantly require more memory than the actual Kafka Producer
> >> buffer.
> >>
> >> I think the safest approach is actually to only store the fixed-size
> >> metadata for the ProductionExceptionHandler.handle:
> >> topic/partition/offset/processorNodeId/taskId, it might be confusing
> >> for the user, but 1) it is still better than nowaday where there are
> >> no context information at all, 2) it would be clearly stated in the
> >> javadoc, 3) the rawSourceKey/Value are already nullable (e.g. the
> >> punctuate case). .
> >>
> >> Do you think it would be a suitable design Sophie?
> >>
> >> Cheers,
> >> Damien
> >>
> >> On Sun, 14 Apr 2024 at 21:30, Loic Greffier <loic.greff...@michelin.com>
> >> wrote:
> >>>
> >>> 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>>
> >>>>>>>>>>>
> >>>>>>>>>>
> >>>>>>>>
> >>>>>
> >>
