Re: [DISCUSS] KIP-1033: Add Kafka Streams exception handler for exceptions occuring during processing

["Matthias J. Sax" <mj...@apache.org>]

Thanks for the update.

I am wondering if we should use `ReadOnlyHeaders` instead of 
`ImmutableHeaders` as interface name?

Also, the returned `Header` interface is technically not immutable 
either, because `Header#key()` returns a mutable byte-array... Would we 
need a `ReadOnlyHeader` interface?

If yes, it seems that `ReadOnlyHeaders` should not be a super-interface 
of `Headers` but it would rather be a standalone interface, and a 
wrapper for a `Headers` instance? And `ReadOnlyHeader` would return some 
immutable type instead of `byte[]` for the value()?

An alternative would be to deep-copy the value byte-array what would not 
be free, but given that we are talking about exception handling, it 
would not be on the hot code path, and thus might be acceptable?


The above seems to increase the complexity significantly though. Hence, 
I have seconds thoughts on the immutability question:

Do we really need to worry about mutability after all, because in the 
end, KS runtime won't read the Headers instance after the handler was 
called, and if a user modifies the passed in headers, there won't be any 
actual damage (ie, no side effects)? For this case, it might even be ok 
to also not add `ImmutableHeaders` to begin with?



Sorry for the forth and back (yes, forth and back, because back and 
forth does not make sense -- it's not logical -- just trying to fix 
English :D) as I did bring up the immutability question in the first 
place...



-Matthias

On 4/25/24 5:56 AM, Loic Greffier wrote:
> Hi Matthias,
> 
> I have updated the KIP regarding points 103 and 108.
> 
> 103.
> I have suggested a new `ImmutableHeaders` interface to deal with the
> immutability concern of the headers, which is basically the `Headers`
> interface without the write accesses.
> 
> public interface ImmutableHeaders {
>      Header lastHeader(String key);
>      Iterable<Header> headers(String key);
>      Header[] toArray();
> }
> 
> The `Headers` interface can be updated accordingly:
> 
> public interface Headers extends ImmutableHeaders, Iterable<Header> {
>      //…
> }
> 
> Loïc

Re: [DISCUSS] KIP-1033: Add Kafka Streams exception handler for exceptions occuring during processing

[Damien Gasparina <d....@gmail.com>]

Thanks for the feedback, I think we should keep two separate callbacks
for serialization and error handlers. It makes sense for type safety
(ProducerRecord<byte[], byte[]> vs POJO) and also for backward
compatibility. On top of that, all metadata provided in the #handle
method would need to be held in memory until the producer invokes its
callback, in the future, having two callbacks might avoid confusion as
some metadata might be provided to #handle and not
#handleSerializationException. I do think that one method would be
cleaner but for backward compatibility, type safety and memory
reasons, we should keep two separate callbacks.

As you suggested Sophie, I updated the KIP to introduce a
SerializationExceptionOrigin enum and added the "origin" parameter to
the #handleSerializationException method.

On Sat, 11 May 2024 at 07:30, Sophie Blee-Goldman <so...@responsive.dev> wrote:
>
> Whoops, just noticed there is already a voting thread for this. Hard to
> keep track with all the KIPs going on right now!
>
> In that case I'll just wait for the SerializationExceptionOrigin thing to
> be added and then I'll vote. Should definitely be able to make 3.8 in this
> case :D
>
> On Fri, May 10, 2024 at 10:10 PM Sophie Blee-Goldman <so...@responsive.dev>
> wrote:
>
> > Sounds like we're more or less in agreement here. I think the KIP just
> > needs one small update still, which is the SerializationExceptionOrigin
> > enum.
> >
> > As discussed there are a few options for this and we're all happy to defer
> > to the preference of the KIP authors, but if we keep the KIP as-is with the
> > two separate methods in the ProductionExceptionHandler, then imo it makes
> > the most sense to add the SerializationExceptionOrigin enum to the
> > ProductionExceptionHandler interface and then add an "origin" parameter to
> > the new  #handleSerializationException method. However you decide to do it,
> > I'm personally happy to vote on this KIP once the KIP is updated.
> >
> >  Just FYI the 3.8 KIP freeze is this upcoming Wednesday, so if you guys
> > would like to target 3.8 for this feature, just make sure to update the KIP
> > and kick off a [VOTE] thread by EOD Monday so that you can close the vote
> > by EOD Wednesday (since it has to be open for 72 hours).
> >
> > Thanks again for this sorely needed feature!
> >
> > On Fri, May 10, 2024 at 10:06 AM Bill Bejeck <bb...@gmail.com> wrote:
> >
> >> Great KIP discussion so far by everyone.
> >> At this point, I'm in agreement with the direction and current state of
> >> the
> >> KIP.
> >>
> >> As for having two separate callbacks for the ProductionExceptionHandler,
> >> I'm somewhat split in that I agree with points raised by Sophie and
> >> Matthias with my final
> >> position being to maintain both callbacks.  IMHO, while there are several
> >> things that could go wrong with producing a message, it seems that
> >> serialization exceptions would be the most common, although I don't have
> >> any data to back that opinion up.  But having said that, should the KIP
> >> authors decide otherwise, I would be fine with that approach as well.
> >>
> >> I'm at the point where I'm comfortable voting for this KIP.
> >>
> >> Thanks,
> >> Bill
> >>
> >> On Thu, May 9, 2024 at 4:28 PM Sophie Blee-Goldman <sophie@responsive.dev
> >> >
> >> wrote:
> >>
> >> > The type safety issue is definitely not solved by having two separate
> >> > callbacks. I just think it gets a bit worse by mashing them into one
> >> > method. At least in the plain #handle method you can be sure that the
> >> type
> >> > is ProducerRecord<byte[], byte[]> and in #handleSerialization the type
> >> is
> >> > some POJO.
> >> >
> >> > And in theory you can just embed the mapping of sink topics to
> >> type/Serde
> >> > based on your topology. Or let's say your output record keys & values
> >> are
> >> > all Strings, and you want to print the String representation in your
> >> > handler, rather than the bytes.
> >> > Having a separate callback means knowing you can simply print the
> >> > ProducerRecord's key/value in the #handleSerialization method, and will
> >> > have to use a StringDeserializer to convert the key/value to its String
> >> > form to print it in the #handle method.
> >> >
> >> > Again, I just feel this will be more straightforward and easy for users
> >> to
> >> > use correctly, but am satisfied either way. I'll shut up now and wait
> >> for
> >> > the KIP authors to make a call on this one way or another, and then I'm
> >> > happy to cast my vote
> >> >
> >> > On Thu, May 9, 2024 at 10:15 AM Matthias J. Sax <mj...@apache.org>
> >> wrote:
> >> >
> >> > > Thanks Sophie! Makes it much clearer where you are coming from.
> >> > >
> >> > > About the Type unsafety: isn't this also an issue for the
> >> > > `handleSerialziationException` case, because the handler is used for
> >> all
> >> > > sink topics, and thus key/value types are not really know w/o taking
> >> the
> >> > > sink topic into account? -- So I am not sure if having two handler
> >> > > methods really helps much with regard to type safety?
> >> > >
> >> > > Just want to make this small comment for completeness. Let's hear what
> >> > > others think. Given that we both don't have a strong opinion but just
> >> a
> >> > > personal preference, we should be able to come to a conclusion quickly
> >> > > and get this KIP approved for 3.8 :)
> >> > >
> >> > >
> >> > > -Matthias
> >> > >
> >> > > On 5/8/24 3:12 PM, Sophie Blee-Goldman wrote:
> >> > > > Well I definitely don't feel super strongly about it, and more
> >> > > importantly,
> >> > > > I'm not a user. So I will happily defer to the preference of anyone
> >> who
> >> > > > will actually be using this feature. But  I'll explain my reasoning:
> >> > > >
> >> > > > There *is* a relevant distinction between these two callbacks --
> >> > because
> >> > > > the passed-in record will have a different type depending on
> >> whether it
> >> > > was
> >> > > > a serialization exception or something else. Even if we combined
> >> them
> >> > > into
> >> > > > a single #handle method, users will still end up implementing two
> >> > > distinct
> >> > > > branches depending on whether it was a serialization exception or
> >> not,
> >> > > > since that determines the type of the ProducerRecord passed in.
> >> > > >
> >> > > > Not to mention they'll need to cast it to a ProducerRecord<byte[],
> >> > > byte[]>
> >> > > > when we could have just passed it in as this type via a dedicated
> >> > > callback.
> >> > > > And note that because of the generics, they can't do an instanceof
> >> > check
> >> > > to
> >> > > > make sure that the record type is ProducerRecord<byte[], byte[]> and
> >> > will
> >> > > > have to suppress the "unchecked cast" warning.
> >> > > >
> >> > > > So if we combined the two callbacks, their handler will look
> >> something
> >> > > like
> >> > > > this:
> >> > > >
> >> > > > @SuppressWarnings("unchecked")
> >> > > > public ProductionExceptionHandlerResponse handle(final
> >> > > ErrorHandlerContext
> >> > > > context,
> >> > > > final ProducerRecord<?, ?> record,
> >> > > > final Exception exception) {
> >> > > > if (exception instanceof SerializationException) {
> >> > > > if (exception.origin().equals(KEY)) {
> >> > > > log.error("Failed to serialize key", exception);
> >> > > > } else {
> >> > > > log.error("Failed to serialize value", exception);
> >> > > > }
> >> > > >
> >> > > > } else {
> >> > > > final ProducerRecord<byte[], byte[]> serializedRecord =
> >> > > (ProducerRecord<byte[],
> >> > > > byte[]>) record;
> >> > > > log.error("Failed to produce record with serialized key={} and
> >> > serialized
> >> > > > value={}",
> >> > > > serializedRecord.key(), serializedRecord.value());
> >> > > > }
> >> > > > return ProductionExceptionHandlerResponse.FAIL;
> >> > > > }
> >> > > >
> >> > > > That seems like the most basic case, and it still haswith distinct
> >> > logic
> >> > > > even if they ultimately handle exceptions the same way. And looking
> >> > > forward
> >> > > > to KIP-1034: Dead-letter queues, it seems all the more likely that
> >> the
> >> > > > actual handling response might be different depending on whether
> >> it's a
> >> > > > serialization exception or not: a serialized record can probably be
> >> > > > retried/sent to a DLQ, whereas a record that can't be serialized
> >> should
> >> > > not
> >> > > > (can't, really) be forwarded to a DLQ. So if they're going to have
> >> > > > completely different implementations depending on whether it's a
> >> > > > serialization exception, why not just give them two separate
> >> callbacks?
> >> > > >
> >> > > > And that's all assuming the user is perfectly aware of the different
> >> > > > exception types and their implications for the type of the
> >> > > ProducerRecord.
> >> > > > Many people might just miss the existence of the
> >> > > > RecordSerializationException altogether --
> >> > > > there are already so many different exception types, ESPECIALLY
> >> when it
> >> > > > comes to the Producer. Not to mention they'll need to understand the
> >> > > > nuances of how the ProducerRecord type changes depending on the
> >> type of
> >> > > > exception that's passed in. And on top of all that, they'll need to
> >> > know
> >> > > > that there is metadata stored in the RecordSerializationException
> >> > > regarding
> >> > > > the origin of the error. Whereas if we just passed in the
> >> > > > SerializationExceptionOrigin to a #handlerSerialization callback,
> >> well,
> >> > > > that's pretty impossible to miss.
> >> > > >
> >> > > > That all just seems like a lot for most people to have to
> >> understand to
> >> > > > implement a ProductionExceptionHandler, which imo is not at all an
> >> > > advanced
> >> > > > feature and should be as straightforward and easy to use as
> >> possible.
> >> > > >
> >> > > > Lastly -- I don't think it's quite fair to compare this to the
> >> > > > RecordDeserializationException. We have a dedicated handler that's
> >> just
> >> > > for
> >> > > > deserialization exceptions specifically, hence there's no worry
> >> about
> >> > > users
> >> > > > having to be aware of the different exception types they might have
> >> to
> >> > > deal
> >> > > > with in the DeserializtionExceptionHandler. Whereas serialization
> >> > > > exceptions are just a subset of what might get passed in to the
> >> > > > ProductionExceptionHandler...
> >> > > >
> >> > > > Just explaining my reasoning -- in the end I leave it up to the KIP
> >> > > authors
> >> > > > and anyone who will actually be using this feature in their
> >> > applications
> >> > > :)
> >> > > >
> >> > > >
> >> > > >
> >> > > > On Tue, May 7, 2024 at 8:35 PM Matthias J. Sax <mj...@apache.org>
> >> > wrote:
> >> > > >
> >> > > >> @Loic, yes, what you describe is exactly what I had in mind.
> >> > > >>
> >> > > >>
> >> > > >>
> >> > > >> @Sophie, can you elaborate a little bit?
> >> > > >>
> >> > > >>> First of all, I agree that it makes sense to maintain the two
> >> > separate
> >> > > >>> callbacks for the ProductionExceptionHandler, since one of them is
> >> > > >>> specifically for serialization exceptions while the other is used
> >> for
> >> > > >>> everything/anything else.
> >> > > >>
> >> > > >> What makes a serialization exception special compare to other
> >> errors
> >> > > >> that it's valuable to treat it differently? Why can we put
> >> "everything
> >> > > >> else" into a single bucket? By your train of though, should we not
> >> > split
> >> > > >> out the "everything else" bucket into a different callback method
> >> for
> >> > > >> every different error? If no, why not, but only for serialization
> >> > > errors?
> >> > > >>
> >> > > >>   From what I believe to remember, historically, we added the
> >> > > >> ProductionExceptionHandler, and kinda just missed the serialization
> >> > > >> error case. And later, when we extended the handler we just could
> >> not
> >> > > >> re-use the existing callback as it was typed with `<byte[],
> >> byte[]>`
> >> > and
> >> > > >> it would have been an incompatible change; so it was rather a
> >> > workaround
> >> > > >> to add the second method to then handler, but not really intended
> >> > > design?
> >> > > >>
> >> > > >>
> >> > > >> It's of course only my personal opinion that I believe a single
> >> > callback
> >> > > >> method is simpler/cleaner compared to sticking with two, and adding
> >> > the
> >> > > >> new exception type to make it backward compatible seems worth it.
> >> It
> >> > > >> also kinda introduces the same patter we use elsewhere (cf
> >> KIP-1036)
> >> > > >> what I actually think is an argument for introducing
> >> > > >> `RercordSerializationExcetpion`, to unify user experience across
> >> the
> >> > > board.
> >> > > >>
> >> > > >> Would be great to hear from others about this point. It's not that
> >> I
> >> > > >> strongly object to having two methods, and I would not block this
> >> KIP
> >> > on
> >> > > >> this question.
> >> > > >>
> >> > > >>
> >> > > >>
> >> > > >> -Matthias
> >> > > >>
> >> > > >>
> >> > > >> On 5/7/24 3:40 PM, Sophie Blee-Goldman wrote:
> >> > > >>> First of all, I agree that it makes sense to maintain the two
> >> > separate
> >> > > >>> callbacks for the ProductionExceptionHandler, since one of them is
> >> > > >>> specifically for serialization exceptions while the other is used
> >> for
> >> > > >>> everything/anything else. I also think we can take advantage of
> >> this
> >> > > fact
> >> > > >>> to simplify things a bit and cut down on the amount of new stuff
> >> > added
> >> > > to
> >> > > >>> the API by just adding a parameter to the
> >> > #handleSerializationException
> >> > > >>> callback and use that to pass in the SerializationExceptionOrigin
> >> > enum
> >> > > to
> >> > > >>> distinguish between key vs value. This way we wouldn't need to
> >> > > introduce
> >> > > >>> yet another exception type (the RecordSerializationException)
> >> just to
> >> > > >> pass
> >> > > >>> in this information.
> >> > > >>>
> >> > > >>> Thoughts?
> >> > > >>>
> >> > > >>> On Tue, May 7, 2024 at 8:33 AM Loic Greffier <
> >> > > loic.greffier@michelin.com
> >> > > >>>
> >> > > >>> wrote:
> >> > > >>>
> >> > > >>>> Hi Matthias,
> >> > > >>>>
> >> > > >>>> To sum up with the ProductionExceptionHandler callback methods
> >> (106)
> >> > > >>>> proposed changes.
> >> > > >>>>
> >> > > >>>> A new method ProductionExceptionHandler#handle is added with the
> >> > > >> following
> >> > > >>>> signature:
> >> > > >>>>
> >> > > >>>>> ProductionExceptionHandlerResponse handle(final
> >> ErrorHandlerContext
> >> > > >>>> context, final ProducerRecord<?,?> record, final Exception
> >> > exception);
> >> > > >>>>
> >> > > >>>> The ProducerRecord<?,?> parameter has changed to accept a
> >> serialized
> >> > > or
> >> > > >>>> non-serialized record.
> >> > > >>>> Thus, the new ProductionExceptionHandler#handle method can handle
> >> > both
> >> > > >>>> production exception and serialization exception.
> >> > > >>>>
> >> > > >>>> Both old ProductionExceptionHandler#handle and
> >> > > >>>> ProductionExceptionHandler#handleSerializationException methods
> >> are
> >> > > now
> >> > > >>>> deprecated.
> >> > > >>>> The old ProductionExceptionHandler#handle method gets a default
> >> > > >>>> implementation, so users do not have to implement a deprecated
> >> > method.
> >> > > >>>>
> >> > > >>>> To handle backward compatibility, the new
> >> > > >>>> ProductionExceptionHandler#handle method gets a default
> >> > > implementation.
> >> > > >>>>
> >> > > >>>>> default ProductionExceptionHandlerResponse handle(final
> >> > > >>>> ErrorHandlerContext context, final ProducerRecord<?, ?> record,
> >> > final
> >> > > >>>> Exception exception) {
> >> > > >>>>>     if (exception instanceof RecordSerializationException) {
> >> > > >>>>>         this.handleSerializationException(record,
> >> > > exception.getCause());
> >> > > >>>>>     }
> >> > > >>>>>
> >> > > >>>>>     return handle((ProducerRecord<byte[], byte[]>) record,
> >> > > exception);
> >> > > >>>>> }
> >> > > >>>>
> >> > > >>>> The default implementation either invokes
> >> > > #handleSerializationException
> >> > > >> or
> >> > > >>>> #handle depending on the type of the exception, thus users still
> >> > > >> relying on
> >> > > >>>> deprecated ProductionExceptionHandler#handle
> >> > > >>>> or ProductionExceptionHandler#handleSerializationException custom
> >> > > >>>> implementations won't break.
> >> > > >>>>
> >> > > >>>> The new ProductionExceptionHandler#handle method is now invoked
> >> in
> >> > > case
> >> > > >> of
> >> > > >>>> serialization exception:
> >> > > >>>>
> >> > > >>>>> public <K, V> void send(final String topic, final K key, final V
> >> > > value,
> >> > > >>>> ...) {
> >> > > >>>>>       try {
> >> > > >>>>>           keyBytes = keySerializer.serialize(topic, headers,
> >> key);
> >> > > >>>>>           ...
> >> > > >>>>>       } catch (final ClassCastException exception) {
> >> > > >>>>>         ...
> >> > > >>>>>       } catch (final Exception exception) {
> >> > > >>>>>
> >> > > >>>>>           try {
> >> > > >>>>>               response =
> >> productionExceptionHandler.handle(context,
> >> > > >>>> record, new
> >> > > >> RecordSerializationException(SerializationExceptionOrigin.KEY,
> >> > > >>>> exception));
> >> > > >>>>>           } catch (final Exception e) {
> >> > > >>>>>           ...
> >> > > >>>>>           }
> >> > > >>>>>       }
> >> > > >>>>> }
> >> > > >>>>
> >> > > >>>> To wrap the origin serialization exception and determine whether
> >> it
> >> > > >> comes
> >> > > >>>> from the key or the value, a new exception class is created:
> >> > > >>>>
> >> > > >>>>> public class RecordSerializationException extends
> >> > > >> SerializationException
> >> > > >>>> {
> >> > > >>>>>       public enum SerializationExceptionOrigin {
> >> > > >>>>>           KEY,
> >> > > >>>>>           VALUE
> >> > > >>>>>       }
> >> > > >>>>>
> >> > > >>>>>       public
> >> > > RecordSerializationException(SerializationExceptionOrigin
> >> > > >>>> origin, final Throwable cause);
> >> > > >>>>> }
> >> > > >>>>
> >> > > >>>> Hope it all makes sense,
> >> > > >>>> Loïc
> >> > > >>>>
> >> > > >>>
> >> > > >>
> >> > > >
> >> > >
> >> >
> >>
> >

Re: [DISCUSS] KIP-1033: Add Kafka Streams exception handler for exceptions occuring during processing

[Sophie Blee-Goldman <so...@responsive.dev>]

Whoops, just noticed there is already a voting thread for this. Hard to
keep track with all the KIPs going on right now!

In that case I'll just wait for the SerializationExceptionOrigin thing to
be added and then I'll vote. Should definitely be able to make 3.8 in this
case :D

On Fri, May 10, 2024 at 10:10 PM Sophie Blee-Goldman <so...@responsive.dev>
wrote:

> Sounds like we're more or less in agreement here. I think the KIP just
> needs one small update still, which is the SerializationExceptionOrigin
> enum.
>
> As discussed there are a few options for this and we're all happy to defer
> to the preference of the KIP authors, but if we keep the KIP as-is with the
> two separate methods in the ProductionExceptionHandler, then imo it makes
> the most sense to add the SerializationExceptionOrigin enum to the
> ProductionExceptionHandler interface and then add an "origin" parameter to
> the new  #handleSerializationException method. However you decide to do it,
> I'm personally happy to vote on this KIP once the KIP is updated.
>
>  Just FYI the 3.8 KIP freeze is this upcoming Wednesday, so if you guys
> would like to target 3.8 for this feature, just make sure to update the KIP
> and kick off a [VOTE] thread by EOD Monday so that you can close the vote
> by EOD Wednesday (since it has to be open for 72 hours).
>
> Thanks again for this sorely needed feature!
>
> On Fri, May 10, 2024 at 10:06 AM Bill Bejeck <bb...@gmail.com> wrote:
>
>> Great KIP discussion so far by everyone.
>> At this point, I'm in agreement with the direction and current state of
>> the
>> KIP.
>>
>> As for having two separate callbacks for the ProductionExceptionHandler,
>> I'm somewhat split in that I agree with points raised by Sophie and
>> Matthias with my final
>> position being to maintain both callbacks.  IMHO, while there are several
>> things that could go wrong with producing a message, it seems that
>> serialization exceptions would be the most common, although I don't have
>> any data to back that opinion up.  But having said that, should the KIP
>> authors decide otherwise, I would be fine with that approach as well.
>>
>> I'm at the point where I'm comfortable voting for this KIP.
>>
>> Thanks,
>> Bill
>>
>> On Thu, May 9, 2024 at 4:28 PM Sophie Blee-Goldman <sophie@responsive.dev
>> >
>> wrote:
>>
>> > The type safety issue is definitely not solved by having two separate
>> > callbacks. I just think it gets a bit worse by mashing them into one
>> > method. At least in the plain #handle method you can be sure that the
>> type
>> > is ProducerRecord<byte[], byte[]> and in #handleSerialization the type
>> is
>> > some POJO.
>> >
>> > And in theory you can just embed the mapping of sink topics to
>> type/Serde
>> > based on your topology. Or let's say your output record keys & values
>> are
>> > all Strings, and you want to print the String representation in your
>> > handler, rather than the bytes.
>> > Having a separate callback means knowing you can simply print the
>> > ProducerRecord's key/value in the #handleSerialization method, and will
>> > have to use a StringDeserializer to convert the key/value to its String
>> > form to print it in the #handle method.
>> >
>> > Again, I just feel this will be more straightforward and easy for users
>> to
>> > use correctly, but am satisfied either way. I'll shut up now and wait
>> for
>> > the KIP authors to make a call on this one way or another, and then I'm
>> > happy to cast my vote
>> >
>> > On Thu, May 9, 2024 at 10:15 AM Matthias J. Sax <mj...@apache.org>
>> wrote:
>> >
>> > > Thanks Sophie! Makes it much clearer where you are coming from.
>> > >
>> > > About the Type unsafety: isn't this also an issue for the
>> > > `handleSerialziationException` case, because the handler is used for
>> all
>> > > sink topics, and thus key/value types are not really know w/o taking
>> the
>> > > sink topic into account? -- So I am not sure if having two handler
>> > > methods really helps much with regard to type safety?
>> > >
>> > > Just want to make this small comment for completeness. Let's hear what
>> > > others think. Given that we both don't have a strong opinion but just
>> a
>> > > personal preference, we should be able to come to a conclusion quickly
>> > > and get this KIP approved for 3.8 :)
>> > >
>> > >
>> > > -Matthias
>> > >
>> > > On 5/8/24 3:12 PM, Sophie Blee-Goldman wrote:
>> > > > Well I definitely don't feel super strongly about it, and more
>> > > importantly,
>> > > > I'm not a user. So I will happily defer to the preference of anyone
>> who
>> > > > will actually be using this feature. But  I'll explain my reasoning:
>> > > >
>> > > > There *is* a relevant distinction between these two callbacks --
>> > because
>> > > > the passed-in record will have a different type depending on
>> whether it
>> > > was
>> > > > a serialization exception or something else. Even if we combined
>> them
>> > > into
>> > > > a single #handle method, users will still end up implementing two
>> > > distinct
>> > > > branches depending on whether it was a serialization exception or
>> not,
>> > > > since that determines the type of the ProducerRecord passed in.
>> > > >
>> > > > Not to mention they'll need to cast it to a ProducerRecord<byte[],
>> > > byte[]>
>> > > > when we could have just passed it in as this type via a dedicated
>> > > callback.
>> > > > And note that because of the generics, they can't do an instanceof
>> > check
>> > > to
>> > > > make sure that the record type is ProducerRecord<byte[], byte[]> and
>> > will
>> > > > have to suppress the "unchecked cast" warning.
>> > > >
>> > > > So if we combined the two callbacks, their handler will look
>> something
>> > > like
>> > > > this:
>> > > >
>> > > > @SuppressWarnings("unchecked")
>> > > > public ProductionExceptionHandlerResponse handle(final
>> > > ErrorHandlerContext
>> > > > context,
>> > > > final ProducerRecord<?, ?> record,
>> > > > final Exception exception) {
>> > > > if (exception instanceof SerializationException) {
>> > > > if (exception.origin().equals(KEY)) {
>> > > > log.error("Failed to serialize key", exception);
>> > > > } else {
>> > > > log.error("Failed to serialize value", exception);
>> > > > }
>> > > >
>> > > > } else {
>> > > > final ProducerRecord<byte[], byte[]> serializedRecord =
>> > > (ProducerRecord<byte[],
>> > > > byte[]>) record;
>> > > > log.error("Failed to produce record with serialized key={} and
>> > serialized
>> > > > value={}",
>> > > > serializedRecord.key(), serializedRecord.value());
>> > > > }
>> > > > return ProductionExceptionHandlerResponse.FAIL;
>> > > > }
>> > > >
>> > > > That seems like the most basic case, and it still haswith distinct
>> > logic
>> > > > even if they ultimately handle exceptions the same way. And looking
>> > > forward
>> > > > to KIP-1034: Dead-letter queues, it seems all the more likely that
>> the
>> > > > actual handling response might be different depending on whether
>> it's a
>> > > > serialization exception or not: a serialized record can probably be
>> > > > retried/sent to a DLQ, whereas a record that can't be serialized
>> should
>> > > not
>> > > > (can't, really) be forwarded to a DLQ. So if they're going to have
>> > > > completely different implementations depending on whether it's a
>> > > > serialization exception, why not just give them two separate
>> callbacks?
>> > > >
>> > > > And that's all assuming the user is perfectly aware of the different
>> > > > exception types and their implications for the type of the
>> > > ProducerRecord.
>> > > > Many people might just miss the existence of the
>> > > > RecordSerializationException altogether --
>> > > > there are already so many different exception types, ESPECIALLY
>> when it
>> > > > comes to the Producer. Not to mention they'll need to understand the
>> > > > nuances of how the ProducerRecord type changes depending on the
>> type of
>> > > > exception that's passed in. And on top of all that, they'll need to
>> > know
>> > > > that there is metadata stored in the RecordSerializationException
>> > > regarding
>> > > > the origin of the error. Whereas if we just passed in the
>> > > > SerializationExceptionOrigin to a #handlerSerialization callback,
>> well,
>> > > > that's pretty impossible to miss.
>> > > >
>> > > > That all just seems like a lot for most people to have to
>> understand to
>> > > > implement a ProductionExceptionHandler, which imo is not at all an
>> > > advanced
>> > > > feature and should be as straightforward and easy to use as
>> possible.
>> > > >
>> > > > Lastly -- I don't think it's quite fair to compare this to the
>> > > > RecordDeserializationException. We have a dedicated handler that's
>> just
>> > > for
>> > > > deserialization exceptions specifically, hence there's no worry
>> about
>> > > users
>> > > > having to be aware of the different exception types they might have
>> to
>> > > deal
>> > > > with in the DeserializtionExceptionHandler. Whereas serialization
>> > > > exceptions are just a subset of what might get passed in to the
>> > > > ProductionExceptionHandler...
>> > > >
>> > > > Just explaining my reasoning -- in the end I leave it up to the KIP
>> > > authors
>> > > > and anyone who will actually be using this feature in their
>> > applications
>> > > :)
>> > > >
>> > > >
>> > > >
>> > > > On Tue, May 7, 2024 at 8:35 PM Matthias J. Sax <mj...@apache.org>
>> > wrote:
>> > > >
>> > > >> @Loic, yes, what you describe is exactly what I had in mind.
>> > > >>
>> > > >>
>> > > >>
>> > > >> @Sophie, can you elaborate a little bit?
>> > > >>
>> > > >>> First of all, I agree that it makes sense to maintain the two
>> > separate
>> > > >>> callbacks for the ProductionExceptionHandler, since one of them is
>> > > >>> specifically for serialization exceptions while the other is used
>> for
>> > > >>> everything/anything else.
>> > > >>
>> > > >> What makes a serialization exception special compare to other
>> errors
>> > > >> that it's valuable to treat it differently? Why can we put
>> "everything
>> > > >> else" into a single bucket? By your train of though, should we not
>> > split
>> > > >> out the "everything else" bucket into a different callback method
>> for
>> > > >> every different error? If no, why not, but only for serialization
>> > > errors?
>> > > >>
>> > > >>   From what I believe to remember, historically, we added the
>> > > >> ProductionExceptionHandler, and kinda just missed the serialization
>> > > >> error case. And later, when we extended the handler we just could
>> not
>> > > >> re-use the existing callback as it was typed with `<byte[],
>> byte[]>`
>> > and
>> > > >> it would have been an incompatible change; so it was rather a
>> > workaround
>> > > >> to add the second method to then handler, but not really intended
>> > > design?
>> > > >>
>> > > >>
>> > > >> It's of course only my personal opinion that I believe a single
>> > callback
>> > > >> method is simpler/cleaner compared to sticking with two, and adding
>> > the
>> > > >> new exception type to make it backward compatible seems worth it.
>> It
>> > > >> also kinda introduces the same patter we use elsewhere (cf
>> KIP-1036)
>> > > >> what I actually think is an argument for introducing
>> > > >> `RercordSerializationExcetpion`, to unify user experience across
>> the
>> > > board.
>> > > >>
>> > > >> Would be great to hear from others about this point. It's not that
>> I
>> > > >> strongly object to having two methods, and I would not block this
>> KIP
>> > on
>> > > >> this question.
>> > > >>
>> > > >>
>> > > >>
>> > > >> -Matthias
>> > > >>
>> > > >>
>> > > >> On 5/7/24 3:40 PM, Sophie Blee-Goldman wrote:
>> > > >>> First of all, I agree that it makes sense to maintain the two
>> > separate
>> > > >>> callbacks for the ProductionExceptionHandler, since one of them is
>> > > >>> specifically for serialization exceptions while the other is used
>> for
>> > > >>> everything/anything else. I also think we can take advantage of
>> this
>> > > fact
>> > > >>> to simplify things a bit and cut down on the amount of new stuff
>> > added
>> > > to
>> > > >>> the API by just adding a parameter to the
>> > #handleSerializationException
>> > > >>> callback and use that to pass in the SerializationExceptionOrigin
>> > enum
>> > > to
>> > > >>> distinguish between key vs value. This way we wouldn't need to
>> > > introduce
>> > > >>> yet another exception type (the RecordSerializationException)
>> just to
>> > > >> pass
>> > > >>> in this information.
>> > > >>>
>> > > >>> Thoughts?
>> > > >>>
>> > > >>> On Tue, May 7, 2024 at 8:33 AM Loic Greffier <
>> > > loic.greffier@michelin.com
>> > > >>>
>> > > >>> wrote:
>> > > >>>
>> > > >>>> Hi Matthias,
>> > > >>>>
>> > > >>>> To sum up with the ProductionExceptionHandler callback methods
>> (106)
>> > > >>>> proposed changes.
>> > > >>>>
>> > > >>>> A new method ProductionExceptionHandler#handle is added with the
>> > > >> following
>> > > >>>> signature:
>> > > >>>>
>> > > >>>>> ProductionExceptionHandlerResponse handle(final
>> ErrorHandlerContext
>> > > >>>> context, final ProducerRecord<?,?> record, final Exception
>> > exception);
>> > > >>>>
>> > > >>>> The ProducerRecord<?,?> parameter has changed to accept a
>> serialized
>> > > or
>> > > >>>> non-serialized record.
>> > > >>>> Thus, the new ProductionExceptionHandler#handle method can handle
>> > both
>> > > >>>> production exception and serialization exception.
>> > > >>>>
>> > > >>>> Both old ProductionExceptionHandler#handle and
>> > > >>>> ProductionExceptionHandler#handleSerializationException methods
>> are
>> > > now
>> > > >>>> deprecated.
>> > > >>>> The old ProductionExceptionHandler#handle method gets a default
>> > > >>>> implementation, so users do not have to implement a deprecated
>> > method.
>> > > >>>>
>> > > >>>> To handle backward compatibility, the new
>> > > >>>> ProductionExceptionHandler#handle method gets a default
>> > > implementation.
>> > > >>>>
>> > > >>>>> default ProductionExceptionHandlerResponse handle(final
>> > > >>>> ErrorHandlerContext context, final ProducerRecord<?, ?> record,
>> > final
>> > > >>>> Exception exception) {
>> > > >>>>>     if (exception instanceof RecordSerializationException) {
>> > > >>>>>         this.handleSerializationException(record,
>> > > exception.getCause());
>> > > >>>>>     }
>> > > >>>>>
>> > > >>>>>     return handle((ProducerRecord<byte[], byte[]>) record,
>> > > exception);
>> > > >>>>> }
>> > > >>>>
>> > > >>>> The default implementation either invokes
>> > > #handleSerializationException
>> > > >> or
>> > > >>>> #handle depending on the type of the exception, thus users still
>> > > >> relying on
>> > > >>>> deprecated ProductionExceptionHandler#handle
>> > > >>>> or ProductionExceptionHandler#handleSerializationException custom
>> > > >>>> implementations won't break.
>> > > >>>>
>> > > >>>> The new ProductionExceptionHandler#handle method is now invoked
>> in
>> > > case
>> > > >> of
>> > > >>>> serialization exception:
>> > > >>>>
>> > > >>>>> public <K, V> void send(final String topic, final K key, final V
>> > > value,
>> > > >>>> ...) {
>> > > >>>>>       try {
>> > > >>>>>           keyBytes = keySerializer.serialize(topic, headers,
>> key);
>> > > >>>>>           ...
>> > > >>>>>       } catch (final ClassCastException exception) {
>> > > >>>>>         ...
>> > > >>>>>       } catch (final Exception exception) {
>> > > >>>>>
>> > > >>>>>           try {
>> > > >>>>>               response =
>> productionExceptionHandler.handle(context,
>> > > >>>> record, new
>> > > >> RecordSerializationException(SerializationExceptionOrigin.KEY,
>> > > >>>> exception));
>> > > >>>>>           } catch (final Exception e) {
>> > > >>>>>           ...
>> > > >>>>>           }
>> > > >>>>>       }
>> > > >>>>> }
>> > > >>>>
>> > > >>>> To wrap the origin serialization exception and determine whether
>> it
>> > > >> comes
>> > > >>>> from the key or the value, a new exception class is created:
>> > > >>>>
>> > > >>>>> public class RecordSerializationException extends
>> > > >> SerializationException
>> > > >>>> {
>> > > >>>>>       public enum SerializationExceptionOrigin {
>> > > >>>>>           KEY,
>> > > >>>>>           VALUE
>> > > >>>>>       }
>> > > >>>>>
>> > > >>>>>       public
>> > > RecordSerializationException(SerializationExceptionOrigin
>> > > >>>> origin, final Throwable cause);
>> > > >>>>> }
>> > > >>>>
>> > > >>>> Hope it all makes sense,
>> > > >>>> Loïc
>> > > >>>>
>> > > >>>
>> > > >>
>> > > >
>> > >
>> >
>>
>

Re: [DISCUSS] KIP-1033: Add Kafka Streams exception handler for exceptions occuring during processing

[Sophie Blee-Goldman <so...@responsive.dev>]

Sounds like we're more or less in agreement here. I think the KIP just
needs one small update still, which is the SerializationExceptionOrigin
enum.

As discussed there are a few options for this and we're all happy to defer
to the preference of the KIP authors, but if we keep the KIP as-is with the
two separate methods in the ProductionExceptionHandler, then imo it makes
the most sense to add the SerializationExceptionOrigin enum to the
ProductionExceptionHandler interface and then add an "origin" parameter to
the new  #handleSerializationException method. However you decide to do it,
I'm personally happy to vote on this KIP once the KIP is updated.

 Just FYI the 3.8 KIP freeze is this upcoming Wednesday, so if you guys
would like to target 3.8 for this feature, just make sure to update the KIP
and kick off a [VOTE] thread by EOD Monday so that you can close the vote
by EOD Wednesday (since it has to be open for 72 hours).

Thanks again for this sorely needed feature!

On Fri, May 10, 2024 at 10:06 AM Bill Bejeck <bb...@gmail.com> wrote:

> Great KIP discussion so far by everyone.
> At this point, I'm in agreement with the direction and current state of the
> KIP.
>
> As for having two separate callbacks for the ProductionExceptionHandler,
> I'm somewhat split in that I agree with points raised by Sophie and
> Matthias with my final
> position being to maintain both callbacks.  IMHO, while there are several
> things that could go wrong with producing a message, it seems that
> serialization exceptions would be the most common, although I don't have
> any data to back that opinion up.  But having said that, should the KIP
> authors decide otherwise, I would be fine with that approach as well.
>
> I'm at the point where I'm comfortable voting for this KIP.
>
> Thanks,
> Bill
>
> On Thu, May 9, 2024 at 4:28 PM Sophie Blee-Goldman <so...@responsive.dev>
> wrote:
>
> > The type safety issue is definitely not solved by having two separate
> > callbacks. I just think it gets a bit worse by mashing them into one
> > method. At least in the plain #handle method you can be sure that the
> type
> > is ProducerRecord<byte[], byte[]> and in #handleSerialization the type is
> > some POJO.
> >
> > And in theory you can just embed the mapping of sink topics to type/Serde
> > based on your topology. Or let's say your output record keys & values are
> > all Strings, and you want to print the String representation in your
> > handler, rather than the bytes.
> > Having a separate callback means knowing you can simply print the
> > ProducerRecord's key/value in the #handleSerialization method, and will
> > have to use a StringDeserializer to convert the key/value to its String
> > form to print it in the #handle method.
> >
> > Again, I just feel this will be more straightforward and easy for users
> to
> > use correctly, but am satisfied either way. I'll shut up now and wait for
> > the KIP authors to make a call on this one way or another, and then I'm
> > happy to cast my vote
> >
> > On Thu, May 9, 2024 at 10:15 AM Matthias J. Sax <mj...@apache.org>
> wrote:
> >
> > > Thanks Sophie! Makes it much clearer where you are coming from.
> > >
> > > About the Type unsafety: isn't this also an issue for the
> > > `handleSerialziationException` case, because the handler is used for
> all
> > > sink topics, and thus key/value types are not really know w/o taking
> the
> > > sink topic into account? -- So I am not sure if having two handler
> > > methods really helps much with regard to type safety?
> > >
> > > Just want to make this small comment for completeness. Let's hear what
> > > others think. Given that we both don't have a strong opinion but just a
> > > personal preference, we should be able to come to a conclusion quickly
> > > and get this KIP approved for 3.8 :)
> > >
> > >
> > > -Matthias
> > >
> > > On 5/8/24 3:12 PM, Sophie Blee-Goldman wrote:
> > > > Well I definitely don't feel super strongly about it, and more
> > > importantly,
> > > > I'm not a user. So I will happily defer to the preference of anyone
> who
> > > > will actually be using this feature. But  I'll explain my reasoning:
> > > >
> > > > There *is* a relevant distinction between these two callbacks --
> > because
> > > > the passed-in record will have a different type depending on whether
> it
> > > was
> > > > a serialization exception or something else. Even if we combined them
> > > into
> > > > a single #handle method, users will still end up implementing two
> > > distinct
> > > > branches depending on whether it was a serialization exception or
> not,
> > > > since that determines the type of the ProducerRecord passed in.
> > > >
> > > > Not to mention they'll need to cast it to a ProducerRecord<byte[],
> > > byte[]>
> > > > when we could have just passed it in as this type via a dedicated
> > > callback.
> > > > And note that because of the generics, they can't do an instanceof
> > check
> > > to
> > > > make sure that the record type is ProducerRecord<byte[], byte[]> and
> > will
> > > > have to suppress the "unchecked cast" warning.
> > > >
> > > > So if we combined the two callbacks, their handler will look
> something
> > > like
> > > > this:
> > > >
> > > > @SuppressWarnings("unchecked")
> > > > public ProductionExceptionHandlerResponse handle(final
> > > ErrorHandlerContext
> > > > context,
> > > > final ProducerRecord<?, ?> record,
> > > > final Exception exception) {
> > > > if (exception instanceof SerializationException) {
> > > > if (exception.origin().equals(KEY)) {
> > > > log.error("Failed to serialize key", exception);
> > > > } else {
> > > > log.error("Failed to serialize value", exception);
> > > > }
> > > >
> > > > } else {
> > > > final ProducerRecord<byte[], byte[]> serializedRecord =
> > > (ProducerRecord<byte[],
> > > > byte[]>) record;
> > > > log.error("Failed to produce record with serialized key={} and
> > serialized
> > > > value={}",
> > > > serializedRecord.key(), serializedRecord.value());
> > > > }
> > > > return ProductionExceptionHandlerResponse.FAIL;
> > > > }
> > > >
> > > > That seems like the most basic case, and it still haswith distinct
> > logic
> > > > even if they ultimately handle exceptions the same way. And looking
> > > forward
> > > > to KIP-1034: Dead-letter queues, it seems all the more likely that
> the
> > > > actual handling response might be different depending on whether
> it's a
> > > > serialization exception or not: a serialized record can probably be
> > > > retried/sent to a DLQ, whereas a record that can't be serialized
> should
> > > not
> > > > (can't, really) be forwarded to a DLQ. So if they're going to have
> > > > completely different implementations depending on whether it's a
> > > > serialization exception, why not just give them two separate
> callbacks?
> > > >
> > > > And that's all assuming the user is perfectly aware of the different
> > > > exception types and their implications for the type of the
> > > ProducerRecord.
> > > > Many people might just miss the existence of the
> > > > RecordSerializationException altogether --
> > > > there are already so many different exception types, ESPECIALLY when
> it
> > > > comes to the Producer. Not to mention they'll need to understand the
> > > > nuances of how the ProducerRecord type changes depending on the type
> of
> > > > exception that's passed in. And on top of all that, they'll need to
> > know
> > > > that there is metadata stored in the RecordSerializationException
> > > regarding
> > > > the origin of the error. Whereas if we just passed in the
> > > > SerializationExceptionOrigin to a #handlerSerialization callback,
> well,
> > > > that's pretty impossible to miss.
> > > >
> > > > That all just seems like a lot for most people to have to understand
> to
> > > > implement a ProductionExceptionHandler, which imo is not at all an
> > > advanced
> > > > feature and should be as straightforward and easy to use as possible.
> > > >
> > > > Lastly -- I don't think it's quite fair to compare this to the
> > > > RecordDeserializationException. We have a dedicated handler that's
> just
> > > for
> > > > deserialization exceptions specifically, hence there's no worry about
> > > users
> > > > having to be aware of the different exception types they might have
> to
> > > deal
> > > > with in the DeserializtionExceptionHandler. Whereas serialization
> > > > exceptions are just a subset of what might get passed in to the
> > > > ProductionExceptionHandler...
> > > >
> > > > Just explaining my reasoning -- in the end I leave it up to the KIP
> > > authors
> > > > and anyone who will actually be using this feature in their
> > applications
> > > :)
> > > >
> > > >
> > > >
> > > > On Tue, May 7, 2024 at 8:35 PM Matthias J. Sax <mj...@apache.org>
> > wrote:
> > > >
> > > >> @Loic, yes, what you describe is exactly what I had in mind.
> > > >>
> > > >>
> > > >>
> > > >> @Sophie, can you elaborate a little bit?
> > > >>
> > > >>> First of all, I agree that it makes sense to maintain the two
> > separate
> > > >>> callbacks for the ProductionExceptionHandler, since one of them is
> > > >>> specifically for serialization exceptions while the other is used
> for
> > > >>> everything/anything else.
> > > >>
> > > >> What makes a serialization exception special compare to other errors
> > > >> that it's valuable to treat it differently? Why can we put
> "everything
> > > >> else" into a single bucket? By your train of though, should we not
> > split
> > > >> out the "everything else" bucket into a different callback method
> for
> > > >> every different error? If no, why not, but only for serialization
> > > errors?
> > > >>
> > > >>   From what I believe to remember, historically, we added the
> > > >> ProductionExceptionHandler, and kinda just missed the serialization
> > > >> error case. And later, when we extended the handler we just could
> not
> > > >> re-use the existing callback as it was typed with `<byte[], byte[]>`
> > and
> > > >> it would have been an incompatible change; so it was rather a
> > workaround
> > > >> to add the second method to then handler, but not really intended
> > > design?
> > > >>
> > > >>
> > > >> It's of course only my personal opinion that I believe a single
> > callback
> > > >> method is simpler/cleaner compared to sticking with two, and adding
> > the
> > > >> new exception type to make it backward compatible seems worth it. It
> > > >> also kinda introduces the same patter we use elsewhere (cf KIP-1036)
> > > >> what I actually think is an argument for introducing
> > > >> `RercordSerializationExcetpion`, to unify user experience across the
> > > board.
> > > >>
> > > >> Would be great to hear from others about this point. It's not that I
> > > >> strongly object to having two methods, and I would not block this
> KIP
> > on
> > > >> this question.
> > > >>
> > > >>
> > > >>
> > > >> -Matthias
> > > >>
> > > >>
> > > >> On 5/7/24 3:40 PM, Sophie Blee-Goldman wrote:
> > > >>> First of all, I agree that it makes sense to maintain the two
> > separate
> > > >>> callbacks for the ProductionExceptionHandler, since one of them is
> > > >>> specifically for serialization exceptions while the other is used
> for
> > > >>> everything/anything else. I also think we can take advantage of
> this
> > > fact
> > > >>> to simplify things a bit and cut down on the amount of new stuff
> > added
> > > to
> > > >>> the API by just adding a parameter to the
> > #handleSerializationException
> > > >>> callback and use that to pass in the SerializationExceptionOrigin
> > enum
> > > to
> > > >>> distinguish between key vs value. This way we wouldn't need to
> > > introduce
> > > >>> yet another exception type (the RecordSerializationException) just
> to
> > > >> pass
> > > >>> in this information.
> > > >>>
> > > >>> Thoughts?
> > > >>>
> > > >>> On Tue, May 7, 2024 at 8:33 AM Loic Greffier <
> > > loic.greffier@michelin.com
> > > >>>
> > > >>> wrote:
> > > >>>
> > > >>>> Hi Matthias,
> > > >>>>
> > > >>>> To sum up with the ProductionExceptionHandler callback methods
> (106)
> > > >>>> proposed changes.
> > > >>>>
> > > >>>> A new method ProductionExceptionHandler#handle is added with the
> > > >> following
> > > >>>> signature:
> > > >>>>
> > > >>>>> ProductionExceptionHandlerResponse handle(final
> ErrorHandlerContext
> > > >>>> context, final ProducerRecord<?,?> record, final Exception
> > exception);
> > > >>>>
> > > >>>> The ProducerRecord<?,?> parameter has changed to accept a
> serialized
> > > or
> > > >>>> non-serialized record.
> > > >>>> Thus, the new ProductionExceptionHandler#handle method can handle
> > both
> > > >>>> production exception and serialization exception.
> > > >>>>
> > > >>>> Both old ProductionExceptionHandler#handle and
> > > >>>> ProductionExceptionHandler#handleSerializationException methods
> are
> > > now
> > > >>>> deprecated.
> > > >>>> The old ProductionExceptionHandler#handle method gets a default
> > > >>>> implementation, so users do not have to implement a deprecated
> > method.
> > > >>>>
> > > >>>> To handle backward compatibility, the new
> > > >>>> ProductionExceptionHandler#handle method gets a default
> > > implementation.
> > > >>>>
> > > >>>>> default ProductionExceptionHandlerResponse handle(final
> > > >>>> ErrorHandlerContext context, final ProducerRecord<?, ?> record,
> > final
> > > >>>> Exception exception) {
> > > >>>>>     if (exception instanceof RecordSerializationException) {
> > > >>>>>         this.handleSerializationException(record,
> > > exception.getCause());
> > > >>>>>     }
> > > >>>>>
> > > >>>>>     return handle((ProducerRecord<byte[], byte[]>) record,
> > > exception);
> > > >>>>> }
> > > >>>>
> > > >>>> The default implementation either invokes
> > > #handleSerializationException
> > > >> or
> > > >>>> #handle depending on the type of the exception, thus users still
> > > >> relying on
> > > >>>> deprecated ProductionExceptionHandler#handle
> > > >>>> or ProductionExceptionHandler#handleSerializationException custom
> > > >>>> implementations won't break.
> > > >>>>
> > > >>>> The new ProductionExceptionHandler#handle method is now invoked in
> > > case
> > > >> of
> > > >>>> serialization exception:
> > > >>>>
> > > >>>>> public <K, V> void send(final String topic, final K key, final V
> > > value,
> > > >>>> ...) {
> > > >>>>>       try {
> > > >>>>>           keyBytes = keySerializer.serialize(topic, headers,
> key);
> > > >>>>>           ...
> > > >>>>>       } catch (final ClassCastException exception) {
> > > >>>>>         ...
> > > >>>>>       } catch (final Exception exception) {
> > > >>>>>
> > > >>>>>           try {
> > > >>>>>               response =
> productionExceptionHandler.handle(context,
> > > >>>> record, new
> > > >> RecordSerializationException(SerializationExceptionOrigin.KEY,
> > > >>>> exception));
> > > >>>>>           } catch (final Exception e) {
> > > >>>>>           ...
> > > >>>>>           }
> > > >>>>>       }
> > > >>>>> }
> > > >>>>
> > > >>>> To wrap the origin serialization exception and determine whether
> it
> > > >> comes
> > > >>>> from the key or the value, a new exception class is created:
> > > >>>>
> > > >>>>> public class RecordSerializationException extends
> > > >> SerializationException
> > > >>>> {
> > > >>>>>       public enum SerializationExceptionOrigin {
> > > >>>>>           KEY,
> > > >>>>>           VALUE
> > > >>>>>       }
> > > >>>>>
> > > >>>>>       public
> > > RecordSerializationException(SerializationExceptionOrigin
> > > >>>> origin, final Throwable cause);
> > > >>>>> }
> > > >>>>
> > > >>>> Hope it all makes sense,
> > > >>>> Loïc
> > > >>>>
> > > >>>
> > > >>
> > > >
> > >
> >
>

Re: [DISCUSS] KIP-1033: Add Kafka Streams exception handler for exceptions occuring during processing

[Bill Bejeck <bb...@gmail.com>]

Great KIP discussion so far by everyone.
At this point, I'm in agreement with the direction and current state of the
KIP.

As for having two separate callbacks for the ProductionExceptionHandler,
I'm somewhat split in that I agree with points raised by Sophie and
Matthias with my final
position being to maintain both callbacks.  IMHO, while there are several
things that could go wrong with producing a message, it seems that
serialization exceptions would be the most common, although I don't have
any data to back that opinion up.  But having said that, should the KIP
authors decide otherwise, I would be fine with that approach as well.

I'm at the point where I'm comfortable voting for this KIP.

Thanks,
Bill

On Thu, May 9, 2024 at 4:28 PM Sophie Blee-Goldman <so...@responsive.dev>
wrote:

> The type safety issue is definitely not solved by having two separate
> callbacks. I just think it gets a bit worse by mashing them into one
> method. At least in the plain #handle method you can be sure that the type
> is ProducerRecord<byte[], byte[]> and in #handleSerialization the type is
> some POJO.
>
> And in theory you can just embed the mapping of sink topics to type/Serde
> based on your topology. Or let's say your output record keys & values are
> all Strings, and you want to print the String representation in your
> handler, rather than the bytes.
> Having a separate callback means knowing you can simply print the
> ProducerRecord's key/value in the #handleSerialization method, and will
> have to use a StringDeserializer to convert the key/value to its String
> form to print it in the #handle method.
>
> Again, I just feel this will be more straightforward and easy for users to
> use correctly, but am satisfied either way. I'll shut up now and wait for
> the KIP authors to make a call on this one way or another, and then I'm
> happy to cast my vote
>
> On Thu, May 9, 2024 at 10:15 AM Matthias J. Sax <mj...@apache.org> wrote:
>
> > Thanks Sophie! Makes it much clearer where you are coming from.
> >
> > About the Type unsafety: isn't this also an issue for the
> > `handleSerialziationException` case, because the handler is used for all
> > sink topics, and thus key/value types are not really know w/o taking the
> > sink topic into account? -- So I am not sure if having two handler
> > methods really helps much with regard to type safety?
> >
> > Just want to make this small comment for completeness. Let's hear what
> > others think. Given that we both don't have a strong opinion but just a
> > personal preference, we should be able to come to a conclusion quickly
> > and get this KIP approved for 3.8 :)
> >
> >
> > -Matthias
> >
> > On 5/8/24 3:12 PM, Sophie Blee-Goldman wrote:
> > > Well I definitely don't feel super strongly about it, and more
> > importantly,
> > > I'm not a user. So I will happily defer to the preference of anyone who
> > > will actually be using this feature. But  I'll explain my reasoning:
> > >
> > > There *is* a relevant distinction between these two callbacks --
> because
> > > the passed-in record will have a different type depending on whether it
> > was
> > > a serialization exception or something else. Even if we combined them
> > into
> > > a single #handle method, users will still end up implementing two
> > distinct
> > > branches depending on whether it was a serialization exception or not,
> > > since that determines the type of the ProducerRecord passed in.
> > >
> > > Not to mention they'll need to cast it to a ProducerRecord<byte[],
> > byte[]>
> > > when we could have just passed it in as this type via a dedicated
> > callback.
> > > And note that because of the generics, they can't do an instanceof
> check
> > to
> > > make sure that the record type is ProducerRecord<byte[], byte[]> and
> will
> > > have to suppress the "unchecked cast" warning.
> > >
> > > So if we combined the two callbacks, their handler will look something
> > like
> > > this:
> > >
> > > @SuppressWarnings("unchecked")
> > > public ProductionExceptionHandlerResponse handle(final
> > ErrorHandlerContext
> > > context,
> > > final ProducerRecord<?, ?> record,
> > > final Exception exception) {
> > > if (exception instanceof SerializationException) {
> > > if (exception.origin().equals(KEY)) {
> > > log.error("Failed to serialize key", exception);
> > > } else {
> > > log.error("Failed to serialize value", exception);
> > > }
> > >
> > > } else {
> > > final ProducerRecord<byte[], byte[]> serializedRecord =
> > (ProducerRecord<byte[],
> > > byte[]>) record;
> > > log.error("Failed to produce record with serialized key={} and
> serialized
> > > value={}",
> > > serializedRecord.key(), serializedRecord.value());
> > > }
> > > return ProductionExceptionHandlerResponse.FAIL;
> > > }
> > >
> > > That seems like the most basic case, and it still haswith distinct
> logic
> > > even if they ultimately handle exceptions the same way. And looking
> > forward
> > > to KIP-1034: Dead-letter queues, it seems all the more likely that the
> > > actual handling response might be different depending on whether it's a
> > > serialization exception or not: a serialized record can probably be
> > > retried/sent to a DLQ, whereas a record that can't be serialized should
> > not
> > > (can't, really) be forwarded to a DLQ. So if they're going to have
> > > completely different implementations depending on whether it's a
> > > serialization exception, why not just give them two separate callbacks?
> > >
> > > And that's all assuming the user is perfectly aware of the different
> > > exception types and their implications for the type of the
> > ProducerRecord.
> > > Many people might just miss the existence of the
> > > RecordSerializationException altogether --
> > > there are already so many different exception types, ESPECIALLY when it
> > > comes to the Producer. Not to mention they'll need to understand the
> > > nuances of how the ProducerRecord type changes depending on the type of
> > > exception that's passed in. And on top of all that, they'll need to
> know
> > > that there is metadata stored in the RecordSerializationException
> > regarding
> > > the origin of the error. Whereas if we just passed in the
> > > SerializationExceptionOrigin to a #handlerSerialization callback, well,
> > > that's pretty impossible to miss.
> > >
> > > That all just seems like a lot for most people to have to understand to
> > > implement a ProductionExceptionHandler, which imo is not at all an
> > advanced
> > > feature and should be as straightforward and easy to use as possible.
> > >
> > > Lastly -- I don't think it's quite fair to compare this to the
> > > RecordDeserializationException. We have a dedicated handler that's just
> > for
> > > deserialization exceptions specifically, hence there's no worry about
> > users
> > > having to be aware of the different exception types they might have to
> > deal
> > > with in the DeserializtionExceptionHandler. Whereas serialization
> > > exceptions are just a subset of what might get passed in to the
> > > ProductionExceptionHandler...
> > >
> > > Just explaining my reasoning -- in the end I leave it up to the KIP
> > authors
> > > and anyone who will actually be using this feature in their
> applications
> > :)
> > >
> > >
> > >
> > > On Tue, May 7, 2024 at 8:35 PM Matthias J. Sax <mj...@apache.org>
> wrote:
> > >
> > >> @Loic, yes, what you describe is exactly what I had in mind.
> > >>
> > >>
> > >>
> > >> @Sophie, can you elaborate a little bit?
> > >>
> > >>> First of all, I agree that it makes sense to maintain the two
> separate
> > >>> callbacks for the ProductionExceptionHandler, since one of them is
> > >>> specifically for serialization exceptions while the other is used for
> > >>> everything/anything else.
> > >>
> > >> What makes a serialization exception special compare to other errors
> > >> that it's valuable to treat it differently? Why can we put "everything
> > >> else" into a single bucket? By your train of though, should we not
> split
> > >> out the "everything else" bucket into a different callback method for
> > >> every different error? If no, why not, but only for serialization
> > errors?
> > >>
> > >>   From what I believe to remember, historically, we added the
> > >> ProductionExceptionHandler, and kinda just missed the serialization
> > >> error case. And later, when we extended the handler we just could not
> > >> re-use the existing callback as it was typed with `<byte[], byte[]>`
> and
> > >> it would have been an incompatible change; so it was rather a
> workaround
> > >> to add the second method to then handler, but not really intended
> > design?
> > >>
> > >>
> > >> It's of course only my personal opinion that I believe a single
> callback
> > >> method is simpler/cleaner compared to sticking with two, and adding
> the
> > >> new exception type to make it backward compatible seems worth it. It
> > >> also kinda introduces the same patter we use elsewhere (cf KIP-1036)
> > >> what I actually think is an argument for introducing
> > >> `RercordSerializationExcetpion`, to unify user experience across the
> > board.
> > >>
> > >> Would be great to hear from others about this point. It's not that I
> > >> strongly object to having two methods, and I would not block this KIP
> on
> > >> this question.
> > >>
> > >>
> > >>
> > >> -Matthias
> > >>
> > >>
> > >> On 5/7/24 3:40 PM, Sophie Blee-Goldman wrote:
> > >>> First of all, I agree that it makes sense to maintain the two
> separate
> > >>> callbacks for the ProductionExceptionHandler, since one of them is
> > >>> specifically for serialization exceptions while the other is used for
> > >>> everything/anything else. I also think we can take advantage of this
> > fact
> > >>> to simplify things a bit and cut down on the amount of new stuff
> added
> > to
> > >>> the API by just adding a parameter to the
> #handleSerializationException
> > >>> callback and use that to pass in the SerializationExceptionOrigin
> enum
> > to
> > >>> distinguish between key vs value. This way we wouldn't need to
> > introduce
> > >>> yet another exception type (the RecordSerializationException) just to
> > >> pass
> > >>> in this information.
> > >>>
> > >>> Thoughts?
> > >>>
> > >>> On Tue, May 7, 2024 at 8:33 AM Loic Greffier <
> > loic.greffier@michelin.com
> > >>>
> > >>> wrote:
> > >>>
> > >>>> Hi Matthias,
> > >>>>
> > >>>> To sum up with the ProductionExceptionHandler callback methods (106)
> > >>>> proposed changes.
> > >>>>
> > >>>> A new method ProductionExceptionHandler#handle is added with the
> > >> following
> > >>>> signature:
> > >>>>
> > >>>>> ProductionExceptionHandlerResponse handle(final ErrorHandlerContext
> > >>>> context, final ProducerRecord<?,?> record, final Exception
> exception);
> > >>>>
> > >>>> The ProducerRecord<?,?> parameter has changed to accept a serialized
> > or
> > >>>> non-serialized record.
> > >>>> Thus, the new ProductionExceptionHandler#handle method can handle
> both
> > >>>> production exception and serialization exception.
> > >>>>
> > >>>> Both old ProductionExceptionHandler#handle and
> > >>>> ProductionExceptionHandler#handleSerializationException methods are
> > now
> > >>>> deprecated.
> > >>>> The old ProductionExceptionHandler#handle method gets a default
> > >>>> implementation, so users do not have to implement a deprecated
> method.
> > >>>>
> > >>>> To handle backward compatibility, the new
> > >>>> ProductionExceptionHandler#handle method gets a default
> > implementation.
> > >>>>
> > >>>>> default ProductionExceptionHandlerResponse handle(final
> > >>>> ErrorHandlerContext context, final ProducerRecord<?, ?> record,
> final
> > >>>> Exception exception) {
> > >>>>>     if (exception instanceof RecordSerializationException) {
> > >>>>>         this.handleSerializationException(record,
> > exception.getCause());
> > >>>>>     }
> > >>>>>
> > >>>>>     return handle((ProducerRecord<byte[], byte[]>) record,
> > exception);
> > >>>>> }
> > >>>>
> > >>>> The default implementation either invokes
> > #handleSerializationException
> > >> or
> > >>>> #handle depending on the type of the exception, thus users still
> > >> relying on
> > >>>> deprecated ProductionExceptionHandler#handle
> > >>>> or ProductionExceptionHandler#handleSerializationException custom
> > >>>> implementations won't break.
> > >>>>
> > >>>> The new ProductionExceptionHandler#handle method is now invoked in
> > case
> > >> of
> > >>>> serialization exception:
> > >>>>
> > >>>>> public <K, V> void send(final String topic, final K key, final V
> > value,
> > >>>> ...) {
> > >>>>>       try {
> > >>>>>           keyBytes = keySerializer.serialize(topic, headers, key);
> > >>>>>           ...
> > >>>>>       } catch (final ClassCastException exception) {
> > >>>>>         ...
> > >>>>>       } catch (final Exception exception) {
> > >>>>>
> > >>>>>           try {
> > >>>>>               response = productionExceptionHandler.handle(context,
> > >>>> record, new
> > >> RecordSerializationException(SerializationExceptionOrigin.KEY,
> > >>>> exception));
> > >>>>>           } catch (final Exception e) {
> > >>>>>           ...
> > >>>>>           }
> > >>>>>       }
> > >>>>> }
> > >>>>
> > >>>> To wrap the origin serialization exception and determine whether it
> > >> comes
> > >>>> from the key or the value, a new exception class is created:
> > >>>>
> > >>>>> public class RecordSerializationException extends
> > >> SerializationException
> > >>>> {
> > >>>>>       public enum SerializationExceptionOrigin {
> > >>>>>           KEY,
> > >>>>>           VALUE
> > >>>>>       }
> > >>>>>
> > >>>>>       public
> > RecordSerializationException(SerializationExceptionOrigin
> > >>>> origin, final Throwable cause);
> > >>>>> }
> > >>>>
> > >>>> Hope it all makes sense,
> > >>>> Loïc
> > >>>>
> > >>>
> > >>
> > >
> >
>

Re: [DISCUSS] KIP-1033: Add Kafka Streams exception handler for exceptions occuring during processing

[Sophie Blee-Goldman <so...@responsive.dev>]

The type safety issue is definitely not solved by having two separate
callbacks. I just think it gets a bit worse by mashing them into one
method. At least in the plain #handle method you can be sure that the type
is ProducerRecord<byte[], byte[]> and in #handleSerialization the type is
some POJO.

And in theory you can just embed the mapping of sink topics to type/Serde
based on your topology. Or let's say your output record keys & values are
all Strings, and you want to print the String representation in your
handler, rather than the bytes.
Having a separate callback means knowing you can simply print the
ProducerRecord's key/value in the #handleSerialization method, and will
have to use a StringDeserializer to convert the key/value to its String
form to print it in the #handle method.

Again, I just feel this will be more straightforward and easy for users to
use correctly, but am satisfied either way. I'll shut up now and wait for
the KIP authors to make a call on this one way or another, and then I'm
happy to cast my vote

On Thu, May 9, 2024 at 10:15 AM Matthias J. Sax <mj...@apache.org> wrote:

> Thanks Sophie! Makes it much clearer where you are coming from.
>
> About the Type unsafety: isn't this also an issue for the
> `handleSerialziationException` case, because the handler is used for all
> sink topics, and thus key/value types are not really know w/o taking the
> sink topic into account? -- So I am not sure if having two handler
> methods really helps much with regard to type safety?
>
> Just want to make this small comment for completeness. Let's hear what
> others think. Given that we both don't have a strong opinion but just a
> personal preference, we should be able to come to a conclusion quickly
> and get this KIP approved for 3.8 :)
>
>
> -Matthias
>
> On 5/8/24 3:12 PM, Sophie Blee-Goldman wrote:
> > Well I definitely don't feel super strongly about it, and more
> importantly,
> > I'm not a user. So I will happily defer to the preference of anyone who
> > will actually be using this feature. But  I'll explain my reasoning:
> >
> > There *is* a relevant distinction between these two callbacks -- because
> > the passed-in record will have a different type depending on whether it
> was
> > a serialization exception or something else. Even if we combined them
> into
> > a single #handle method, users will still end up implementing two
> distinct
> > branches depending on whether it was a serialization exception or not,
> > since that determines the type of the ProducerRecord passed in.
> >
> > Not to mention they'll need to cast it to a ProducerRecord<byte[],
> byte[]>
> > when we could have just passed it in as this type via a dedicated
> callback.
> > And note that because of the generics, they can't do an instanceof check
> to
> > make sure that the record type is ProducerRecord<byte[], byte[]> and will
> > have to suppress the "unchecked cast" warning.
> >
> > So if we combined the two callbacks, their handler will look something
> like
> > this:
> >
> > @SuppressWarnings("unchecked")
> > public ProductionExceptionHandlerResponse handle(final
> ErrorHandlerContext
> > context,
> > final ProducerRecord<?, ?> record,
> > final Exception exception) {
> > if (exception instanceof SerializationException) {
> > if (exception.origin().equals(KEY)) {
> > log.error("Failed to serialize key", exception);
> > } else {
> > log.error("Failed to serialize value", exception);
> > }
> >
> > } else {
> > final ProducerRecord<byte[], byte[]> serializedRecord =
> (ProducerRecord<byte[],
> > byte[]>) record;
> > log.error("Failed to produce record with serialized key={} and serialized
> > value={}",
> > serializedRecord.key(), serializedRecord.value());
> > }
> > return ProductionExceptionHandlerResponse.FAIL;
> > }
> >
> > That seems like the most basic case, and it still haswith distinct logic
> > even if they ultimately handle exceptions the same way. And looking
> forward
> > to KIP-1034: Dead-letter queues, it seems all the more likely that the
> > actual handling response might be different depending on whether it's a
> > serialization exception or not: a serialized record can probably be
> > retried/sent to a DLQ, whereas a record that can't be serialized should
> not
> > (can't, really) be forwarded to a DLQ. So if they're going to have
> > completely different implementations depending on whether it's a
> > serialization exception, why not just give them two separate callbacks?
> >
> > And that's all assuming the user is perfectly aware of the different
> > exception types and their implications for the type of the
> ProducerRecord.
> > Many people might just miss the existence of the
> > RecordSerializationException altogether --
> > there are already so many different exception types, ESPECIALLY when it
> > comes to the Producer. Not to mention they'll need to understand the
> > nuances of how the ProducerRecord type changes depending on the type of
> > exception that's passed in. And on top of all that, they'll need to know
> > that there is metadata stored in the RecordSerializationException
> regarding
> > the origin of the error. Whereas if we just passed in the
> > SerializationExceptionOrigin to a #handlerSerialization callback, well,
> > that's pretty impossible to miss.
> >
> > That all just seems like a lot for most people to have to understand to
> > implement a ProductionExceptionHandler, which imo is not at all an
> advanced
> > feature and should be as straightforward and easy to use as possible.
> >
> > Lastly -- I don't think it's quite fair to compare this to the
> > RecordDeserializationException. We have a dedicated handler that's just
> for
> > deserialization exceptions specifically, hence there's no worry about
> users
> > having to be aware of the different exception types they might have to
> deal
> > with in the DeserializtionExceptionHandler. Whereas serialization
> > exceptions are just a subset of what might get passed in to the
> > ProductionExceptionHandler...
> >
> > Just explaining my reasoning -- in the end I leave it up to the KIP
> authors
> > and anyone who will actually be using this feature in their applications
> :)
> >
> >
> >
> > On Tue, May 7, 2024 at 8:35 PM Matthias J. Sax <mj...@apache.org> wrote:
> >
> >> @Loic, yes, what you describe is exactly what I had in mind.
> >>
> >>
> >>
> >> @Sophie, can you elaborate a little bit?
> >>
> >>> First of all, I agree that it makes sense to maintain the two separate
> >>> callbacks for the ProductionExceptionHandler, since one of them is
> >>> specifically for serialization exceptions while the other is used for
> >>> everything/anything else.
> >>
> >> What makes a serialization exception special compare to other errors
> >> that it's valuable to treat it differently? Why can we put "everything
> >> else" into a single bucket? By your train of though, should we not split
> >> out the "everything else" bucket into a different callback method for
> >> every different error? If no, why not, but only for serialization
> errors?
> >>
> >>   From what I believe to remember, historically, we added the
> >> ProductionExceptionHandler, and kinda just missed the serialization
> >> error case. And later, when we extended the handler we just could not
> >> re-use the existing callback as it was typed with `<byte[], byte[]>` and
> >> it would have been an incompatible change; so it was rather a workaround
> >> to add the second method to then handler, but not really intended
> design?
> >>
> >>
> >> It's of course only my personal opinion that I believe a single callback
> >> method is simpler/cleaner compared to sticking with two, and adding the
> >> new exception type to make it backward compatible seems worth it. It
> >> also kinda introduces the same patter we use elsewhere (cf KIP-1036)
> >> what I actually think is an argument for introducing
> >> `RercordSerializationExcetpion`, to unify user experience across the
> board.
> >>
> >> Would be great to hear from others about this point. It's not that I
> >> strongly object to having two methods, and I would not block this KIP on
> >> this question.
> >>
> >>
> >>
> >> -Matthias
> >>
> >>
> >> On 5/7/24 3:40 PM, Sophie Blee-Goldman wrote:
> >>> First of all, I agree that it makes sense to maintain the two separate
> >>> callbacks for the ProductionExceptionHandler, since one of them is
> >>> specifically for serialization exceptions while the other is used for
> >>> everything/anything else. I also think we can take advantage of this
> fact
> >>> to simplify things a bit and cut down on the amount of new stuff added
> to
> >>> the API by just adding a parameter to the #handleSerializationException
> >>> callback and use that to pass in the SerializationExceptionOrigin enum
> to
> >>> distinguish between key vs value. This way we wouldn't need to
> introduce
> >>> yet another exception type (the RecordSerializationException) just to
> >> pass
> >>> in this information.
> >>>
> >>> Thoughts?
> >>>
> >>> On Tue, May 7, 2024 at 8:33 AM Loic Greffier <
> loic.greffier@michelin.com
> >>>
> >>> wrote:
> >>>
> >>>> Hi Matthias,
> >>>>
> >>>> To sum up with the ProductionExceptionHandler callback methods (106)
> >>>> proposed changes.
> >>>>
> >>>> A new method ProductionExceptionHandler#handle is added with the
> >> following
> >>>> signature:
> >>>>
> >>>>> ProductionExceptionHandlerResponse handle(final ErrorHandlerContext
> >>>> context, final ProducerRecord<?,?> record, final Exception exception);
> >>>>
> >>>> The ProducerRecord<?,?> parameter has changed to accept a serialized
> or
> >>>> non-serialized record.
> >>>> Thus, the new ProductionExceptionHandler#handle method can handle both
> >>>> production exception and serialization exception.
> >>>>
> >>>> Both old ProductionExceptionHandler#handle and
> >>>> ProductionExceptionHandler#handleSerializationException methods are
> now
> >>>> deprecated.
> >>>> The old ProductionExceptionHandler#handle method gets a default
> >>>> implementation, so users do not have to implement a deprecated method.
> >>>>
> >>>> To handle backward compatibility, the new
> >>>> ProductionExceptionHandler#handle method gets a default
> implementation.
> >>>>
> >>>>> default ProductionExceptionHandlerResponse handle(final
> >>>> ErrorHandlerContext context, final ProducerRecord<?, ?> record, final
> >>>> Exception exception) {
> >>>>>     if (exception instanceof RecordSerializationException) {
> >>>>>         this.handleSerializationException(record,
> exception.getCause());
> >>>>>     }
> >>>>>
> >>>>>     return handle((ProducerRecord<byte[], byte[]>) record,
> exception);
> >>>>> }
> >>>>
> >>>> The default implementation either invokes
> #handleSerializationException
> >> or
> >>>> #handle depending on the type of the exception, thus users still
> >> relying on
> >>>> deprecated ProductionExceptionHandler#handle
> >>>> or ProductionExceptionHandler#handleSerializationException custom
> >>>> implementations won't break.
> >>>>
> >>>> The new ProductionExceptionHandler#handle method is now invoked in
> case
> >> of
> >>>> serialization exception:
> >>>>
> >>>>> public <K, V> void send(final String topic, final K key, final V
> value,
> >>>> ...) {
> >>>>>       try {
> >>>>>           keyBytes = keySerializer.serialize(topic, headers, key);
> >>>>>           ...
> >>>>>       } catch (final ClassCastException exception) {
> >>>>>         ...
> >>>>>       } catch (final Exception exception) {
> >>>>>
> >>>>>           try {
> >>>>>               response = productionExceptionHandler.handle(context,
> >>>> record, new
> >> RecordSerializationException(SerializationExceptionOrigin.KEY,
> >>>> exception));
> >>>>>           } catch (final Exception e) {
> >>>>>           ...
> >>>>>           }
> >>>>>       }
> >>>>> }
> >>>>
> >>>> To wrap the origin serialization exception and determine whether it
> >> comes
> >>>> from the key or the value, a new exception class is created:
> >>>>
> >>>>> public class RecordSerializationException extends
> >> SerializationException
> >>>> {
> >>>>>       public enum SerializationExceptionOrigin {
> >>>>>           KEY,
> >>>>>           VALUE
> >>>>>       }
> >>>>>
> >>>>>       public
> RecordSerializationException(SerializationExceptionOrigin
> >>>> origin, final Throwable cause);
> >>>>> }
> >>>>
> >>>> Hope it all makes sense,
> >>>> Loïc
> >>>>
> >>>
> >>
> >
>

Re: [DISCUSS] KIP-1033: Add Kafka Streams exception handler for exceptions occuring during processing

["Matthias J. Sax" <mj...@apache.org>]

Thanks Sophie! Makes it much clearer where you are coming from.

About the Type unsafety: isn't this also an issue for the 
`handleSerialziationException` case, because the handler is used for all 
sink topics, and thus key/value types are not really know w/o taking the 
sink topic into account? -- So I am not sure if having two handler 
methods really helps much with regard to type safety?

Just want to make this small comment for completeness. Let's hear what 
others think. Given that we both don't have a strong opinion but just a 
personal preference, we should be able to come to a conclusion quickly 
and get this KIP approved for 3.8 :)


-Matthias

On 5/8/24 3:12 PM, Sophie Blee-Goldman wrote:
> Well I definitely don't feel super strongly about it, and more importantly,
> I'm not a user. So I will happily defer to the preference of anyone who
> will actually be using this feature. But  I'll explain my reasoning:
> 
> There *is* a relevant distinction between these two callbacks -- because
> the passed-in record will have a different type depending on whether it was
> a serialization exception or something else. Even if we combined them into
> a single #handle method, users will still end up implementing two distinct
> branches depending on whether it was a serialization exception or not,
> since that determines the type of the ProducerRecord passed in.
> 
> Not to mention they'll need to cast it to a ProducerRecord<byte[], byte[]>
> when we could have just passed it in as this type via a dedicated callback.
> And note that because of the generics, they can't do an instanceof check to
> make sure that the record type is ProducerRecord<byte[], byte[]> and will
> have to suppress the "unchecked cast" warning.
> 
> So if we combined the two callbacks, their handler will look something like
> this:
> 
> @SuppressWarnings("unchecked")
> public ProductionExceptionHandlerResponse handle(final ErrorHandlerContext
> context,
> final ProducerRecord<?, ?> record,
> final Exception exception) {
> if (exception instanceof SerializationException) {
> if (exception.origin().equals(KEY)) {
> log.error("Failed to serialize key", exception);
> } else {
> log.error("Failed to serialize value", exception);
> }
> 
> } else {
> final ProducerRecord<byte[], byte[]> serializedRecord = (ProducerRecord<byte[],
> byte[]>) record;
> log.error("Failed to produce record with serialized key={} and serialized
> value={}",
> serializedRecord.key(), serializedRecord.value());
> }
> return ProductionExceptionHandlerResponse.FAIL;
> }
> 
> That seems like the most basic case, and it still haswith distinct logic
> even if they ultimately handle exceptions the same way. And looking forward
> to KIP-1034: Dead-letter queues, it seems all the more likely that the
> actual handling response might be different depending on whether it's a
> serialization exception or not: a serialized record can probably be
> retried/sent to a DLQ, whereas a record that can't be serialized should not
> (can't, really) be forwarded to a DLQ. So if they're going to have
> completely different implementations depending on whether it's a
> serialization exception, why not just give them two separate callbacks?
> 
> And that's all assuming the user is perfectly aware of the different
> exception types and their implications for the type of the ProducerRecord.
> Many people might just miss the existence of the
> RecordSerializationException altogether --
> there are already so many different exception types, ESPECIALLY when it
> comes to the Producer. Not to mention they'll need to understand the
> nuances of how the ProducerRecord type changes depending on the type of
> exception that's passed in. And on top of all that, they'll need to know
> that there is metadata stored in the RecordSerializationException regarding
> the origin of the error. Whereas if we just passed in the
> SerializationExceptionOrigin to a #handlerSerialization callback, well,
> that's pretty impossible to miss.
> 
> That all just seems like a lot for most people to have to understand to
> implement a ProductionExceptionHandler, which imo is not at all an advanced
> feature and should be as straightforward and easy to use as possible.
> 
> Lastly -- I don't think it's quite fair to compare this to the
> RecordDeserializationException. We have a dedicated handler that's just for
> deserialization exceptions specifically, hence there's no worry about users
> having to be aware of the different exception types they might have to deal
> with in the DeserializtionExceptionHandler. Whereas serialization
> exceptions are just a subset of what might get passed in to the
> ProductionExceptionHandler...
> 
> Just explaining my reasoning -- in the end I leave it up to the KIP authors
> and anyone who will actually be using this feature in their applications :)
> 
> 
> 
> On Tue, May 7, 2024 at 8:35 PM Matthias J. Sax <mj...@apache.org> wrote:
> 
>> @Loic, yes, what you describe is exactly what I had in mind.
>>
>>
>>
>> @Sophie, can you elaborate a little bit?
>>
>>> First of all, I agree that it makes sense to maintain the two separate
>>> callbacks for the ProductionExceptionHandler, since one of them is
>>> specifically for serialization exceptions while the other is used for
>>> everything/anything else.
>>
>> What makes a serialization exception special compare to other errors
>> that it's valuable to treat it differently? Why can we put "everything
>> else" into a single bucket? By your train of though, should we not split
>> out the "everything else" bucket into a different callback method for
>> every different error? If no, why not, but only for serialization errors?
>>
>>   From what I believe to remember, historically, we added the
>> ProductionExceptionHandler, and kinda just missed the serialization
>> error case. And later, when we extended the handler we just could not
>> re-use the existing callback as it was typed with `<byte[], byte[]>` and
>> it would have been an incompatible change; so it was rather a workaround
>> to add the second method to then handler, but not really intended design?
>>
>>
>> It's of course only my personal opinion that I believe a single callback
>> method is simpler/cleaner compared to sticking with two, and adding the
>> new exception type to make it backward compatible seems worth it. It
>> also kinda introduces the same patter we use elsewhere (cf KIP-1036)
>> what I actually think is an argument for introducing
>> `RercordSerializationExcetpion`, to unify user experience across the board.
>>
>> Would be great to hear from others about this point. It's not that I
>> strongly object to having two methods, and I would not block this KIP on
>> this question.
>>
>>
>>
>> -Matthias
>>
>>
>> On 5/7/24 3:40 PM, Sophie Blee-Goldman wrote:
>>> First of all, I agree that it makes sense to maintain the two separate
>>> callbacks for the ProductionExceptionHandler, since one of them is
>>> specifically for serialization exceptions while the other is used for
>>> everything/anything else. I also think we can take advantage of this fact
>>> to simplify things a bit and cut down on the amount of new stuff added to
>>> the API by just adding a parameter to the #handleSerializationException
>>> callback and use that to pass in the SerializationExceptionOrigin enum to
>>> distinguish between key vs value. This way we wouldn't need to introduce
>>> yet another exception type (the RecordSerializationException) just to
>> pass
>>> in this information.
>>>
>>> Thoughts?
>>>
>>> On Tue, May 7, 2024 at 8:33 AM Loic Greffier <loic.greffier@michelin.com
>>>
>>> wrote:
>>>
>>>> Hi Matthias,
>>>>
>>>> To sum up with the ProductionExceptionHandler callback methods (106)
>>>> proposed changes.
>>>>
>>>> A new method ProductionExceptionHandler#handle is added with the
>> following
>>>> signature:
>>>>
>>>>> ProductionExceptionHandlerResponse handle(final ErrorHandlerContext
>>>> context, final ProducerRecord<?,?> record, final Exception exception);
>>>>
>>>> The ProducerRecord<?,?> parameter has changed to accept a serialized or
>>>> non-serialized record.
>>>> Thus, the new ProductionExceptionHandler#handle method can handle both
>>>> production exception and serialization exception.
>>>>
>>>> Both old ProductionExceptionHandler#handle and
>>>> ProductionExceptionHandler#handleSerializationException methods are now
>>>> deprecated.
>>>> The old ProductionExceptionHandler#handle method gets a default
>>>> implementation, so users do not have to implement a deprecated method.
>>>>
>>>> To handle backward compatibility, the new
>>>> ProductionExceptionHandler#handle method gets a default implementation.
>>>>
>>>>> default ProductionExceptionHandlerResponse handle(final
>>>> ErrorHandlerContext context, final ProducerRecord<?, ?> record, final
>>>> Exception exception) {
>>>>>     if (exception instanceof RecordSerializationException) {
>>>>>         this.handleSerializationException(record, exception.getCause());
>>>>>     }
>>>>>
>>>>>     return handle((ProducerRecord<byte[], byte[]>) record, exception);
>>>>> }
>>>>
>>>> The default implementation either invokes #handleSerializationException
>> or
>>>> #handle depending on the type of the exception, thus users still
>> relying on
>>>> deprecated ProductionExceptionHandler#handle
>>>> or ProductionExceptionHandler#handleSerializationException custom
>>>> implementations won't break.
>>>>
>>>> The new ProductionExceptionHandler#handle method is now invoked in case
>> of
>>>> serialization exception:
>>>>
>>>>> public <K, V> void send(final String topic, final K key, final V value,
>>>> ...) {
>>>>>       try {
>>>>>           keyBytes = keySerializer.serialize(topic, headers, key);
>>>>>           ...
>>>>>       } catch (final ClassCastException exception) {
>>>>>         ...
>>>>>       } catch (final Exception exception) {
>>>>>
>>>>>           try {
>>>>>               response = productionExceptionHandler.handle(context,
>>>> record, new
>> RecordSerializationException(SerializationExceptionOrigin.KEY,
>>>> exception));
>>>>>           } catch (final Exception e) {
>>>>>           ...
>>>>>           }
>>>>>       }
>>>>> }
>>>>
>>>> To wrap the origin serialization exception and determine whether it
>> comes
>>>> from the key or the value, a new exception class is created:
>>>>
>>>>> public class RecordSerializationException extends
>> SerializationException
>>>> {
>>>>>       public enum SerializationExceptionOrigin {
>>>>>           KEY,
>>>>>           VALUE
>>>>>       }
>>>>>
>>>>>       public RecordSerializationException(SerializationExceptionOrigin
>>>> origin, final Throwable cause);
>>>>> }
>>>>
>>>> Hope it all makes sense,
>>>> Loïc
>>>>
>>>
>>
> 

Re: [DISCUSS] KIP-1033: Add Kafka Streams exception handler for exceptions occuring during processing

[Sophie Blee-Goldman <so...@responsive.dev>]

Well I definitely don't feel super strongly about it, and more importantly,
I'm not a user. So I will happily defer to the preference of anyone who
will actually be using this feature. But  I'll explain my reasoning:

There *is* a relevant distinction between these two callbacks -- because
the passed-in record will have a different type depending on whether it was
a serialization exception or something else. Even if we combined them into
a single #handle method, users will still end up implementing two distinct
branches depending on whether it was a serialization exception or not,
since that determines the type of the ProducerRecord passed in.

Not to mention they'll need to cast it to a ProducerRecord<byte[], byte[]>
when we could have just passed it in as this type via a dedicated callback.
And note that because of the generics, they can't do an instanceof check to
make sure that the record type is ProducerRecord<byte[], byte[]> and will
have to suppress the "unchecked cast" warning.

So if we combined the two callbacks, their handler will look something like
this:

@SuppressWarnings("unchecked")
public ProductionExceptionHandlerResponse handle(final ErrorHandlerContext
context,
final ProducerRecord<?, ?> record,
final Exception exception) {
if (exception instanceof SerializationException) {
if (exception.origin().equals(KEY)) {
log.error("Failed to serialize key", exception);
} else {
log.error("Failed to serialize value", exception);
}

} else {
final ProducerRecord<byte[], byte[]> serializedRecord = (ProducerRecord<byte[],
byte[]>) record;
log.error("Failed to produce record with serialized key={} and serialized
value={}",
serializedRecord.key(), serializedRecord.value());
}
return ProductionExceptionHandlerResponse.FAIL;
}

That seems like the most basic case, and it still haswith distinct logic
even if they ultimately handle exceptions the same way. And looking forward
to KIP-1034: Dead-letter queues, it seems all the more likely that the
actual handling response might be different depending on whether it's a
serialization exception or not: a serialized record can probably be
retried/sent to a DLQ, whereas a record that can't be serialized should not
(can't, really) be forwarded to a DLQ. So if they're going to have
completely different implementations depending on whether it's a
serialization exception, why not just give them two separate callbacks?

And that's all assuming the user is perfectly aware of the different
exception types and their implications for the type of the ProducerRecord.
Many people might just miss the existence of the
RecordSerializationException altogether --
there are already so many different exception types, ESPECIALLY when it
comes to the Producer. Not to mention they'll need to understand the
nuances of how the ProducerRecord type changes depending on the type of
exception that's passed in. And on top of all that, they'll need to know
that there is metadata stored in the RecordSerializationException regarding
the origin of the error. Whereas if we just passed in the
SerializationExceptionOrigin to a #handlerSerialization callback, well,
that's pretty impossible to miss.

That all just seems like a lot for most people to have to understand to
implement a ProductionExceptionHandler, which imo is not at all an advanced
feature and should be as straightforward and easy to use as possible.

Lastly -- I don't think it's quite fair to compare this to the
RecordDeserializationException. We have a dedicated handler that's just for
deserialization exceptions specifically, hence there's no worry about users
having to be aware of the different exception types they might have to deal
with in the DeserializtionExceptionHandler. Whereas serialization
exceptions are just a subset of what might get passed in to the
ProductionExceptionHandler...

Just explaining my reasoning -- in the end I leave it up to the KIP authors
and anyone who will actually be using this feature in their applications :)



On Tue, May 7, 2024 at 8:35 PM Matthias J. Sax <mj...@apache.org> wrote:

> @Loic, yes, what you describe is exactly what I had in mind.
>
>
>
> @Sophie, can you elaborate a little bit?
>
> > First of all, I agree that it makes sense to maintain the two separate
> > callbacks for the ProductionExceptionHandler, since one of them is
> > specifically for serialization exceptions while the other is used for
> > everything/anything else.
>
> What makes a serialization exception special compare to other errors
> that it's valuable to treat it differently? Why can we put "everything
> else" into a single bucket? By your train of though, should we not split
> out the "everything else" bucket into a different callback method for
> every different error? If no, why not, but only for serialization errors?
>
>  From what I believe to remember, historically, we added the
> ProductionExceptionHandler, and kinda just missed the serialization
> error case. And later, when we extended the handler we just could not
> re-use the existing callback as it was typed with `<byte[], byte[]>` and
> it would have been an incompatible change; so it was rather a workaround
> to add the second method to then handler, but not really intended design?
>
>
> It's of course only my personal opinion that I believe a single callback
> method is simpler/cleaner compared to sticking with two, and adding the
> new exception type to make it backward compatible seems worth it. It
> also kinda introduces the same patter we use elsewhere (cf KIP-1036)
> what I actually think is an argument for introducing
> `RercordSerializationExcetpion`, to unify user experience across the board.
>
> Would be great to hear from others about this point. It's not that I
> strongly object to having two methods, and I would not block this KIP on
> this question.
>
>
>
> -Matthias
>
>
> On 5/7/24 3:40 PM, Sophie Blee-Goldman wrote:
> > First of all, I agree that it makes sense to maintain the two separate
> > callbacks for the ProductionExceptionHandler, since one of them is
> > specifically for serialization exceptions while the other is used for
> > everything/anything else. I also think we can take advantage of this fact
> > to simplify things a bit and cut down on the amount of new stuff added to
> > the API by just adding a parameter to the #handleSerializationException
> > callback and use that to pass in the SerializationExceptionOrigin enum to
> > distinguish between key vs value. This way we wouldn't need to introduce
> > yet another exception type (the RecordSerializationException) just to
> pass
> > in this information.
> >
> > Thoughts?
> >
> > On Tue, May 7, 2024 at 8:33 AM Loic Greffier <loic.greffier@michelin.com
> >
> > wrote:
> >
> >> Hi Matthias,
> >>
> >> To sum up with the ProductionExceptionHandler callback methods (106)
> >> proposed changes.
> >>
> >> A new method ProductionExceptionHandler#handle is added with the
> following
> >> signature:
> >>
> >>> ProductionExceptionHandlerResponse handle(final ErrorHandlerContext
> >> context, final ProducerRecord<?,?> record, final Exception exception);
> >>
> >> The ProducerRecord<?,?> parameter has changed to accept a serialized or
> >> non-serialized record.
> >> Thus, the new ProductionExceptionHandler#handle method can handle both
> >> production exception and serialization exception.
> >>
> >> Both old ProductionExceptionHandler#handle and
> >> ProductionExceptionHandler#handleSerializationException methods are now
> >> deprecated.
> >> The old ProductionExceptionHandler#handle method gets a default
> >> implementation, so users do not have to implement a deprecated method.
> >>
> >> To handle backward compatibility, the new
> >> ProductionExceptionHandler#handle method gets a default implementation.
> >>
> >>> default ProductionExceptionHandlerResponse handle(final
> >> ErrorHandlerContext context, final ProducerRecord<?, ?> record, final
> >> Exception exception) {
> >>>    if (exception instanceof RecordSerializationException) {
> >>>        this.handleSerializationException(record, exception.getCause());
> >>>    }
> >>>
> >>>    return handle((ProducerRecord<byte[], byte[]>) record, exception);
> >>> }
> >>
> >> The default implementation either invokes #handleSerializationException
> or
> >> #handle depending on the type of the exception, thus users still
> relying on
> >> deprecated ProductionExceptionHandler#handle
> >> or ProductionExceptionHandler#handleSerializationException custom
> >> implementations won't break.
> >>
> >> The new ProductionExceptionHandler#handle method is now invoked in case
> of
> >> serialization exception:
> >>
> >>> public <K, V> void send(final String topic, final K key, final V value,
> >> ...) {
> >>>      try {
> >>>          keyBytes = keySerializer.serialize(topic, headers, key);
> >>>          ...
> >>>      } catch (final ClassCastException exception) {
> >>>        ...
> >>>      } catch (final Exception exception) {
> >>>
> >>>          try {
> >>>              response = productionExceptionHandler.handle(context,
> >> record, new
> RecordSerializationException(SerializationExceptionOrigin.KEY,
> >> exception));
> >>>          } catch (final Exception e) {
> >>>          ...
> >>>          }
> >>>      }
> >>> }
> >>
> >> To wrap the origin serialization exception and determine whether it
> comes
> >> from the key or the value, a new exception class is created:
> >>
> >>> public class RecordSerializationException extends
> SerializationException
> >> {
> >>>      public enum SerializationExceptionOrigin {
> >>>          KEY,
> >>>          VALUE
> >>>      }
> >>>
> >>>      public RecordSerializationException(SerializationExceptionOrigin
> >> origin, final Throwable cause);
> >>> }
> >>
> >> Hope it all makes sense,
> >> Loïc
> >>
> >
>

Re: [DISCUSS] KIP-1033: Add Kafka Streams exception handler for exceptions occuring during processing

["Matthias J. Sax" <mj...@apache.org>]

@Loic, yes, what you describe is exactly what I had in mind.



@Sophie, can you elaborate a little bit?

> First of all, I agree that it makes sense to maintain the two separate
> callbacks for the ProductionExceptionHandler, since one of them is
> specifically for serialization exceptions while the other is used for
> everything/anything else.

What makes a serialization exception special compare to other errors 
that it's valuable to treat it differently? Why can we put "everything 
else" into a single bucket? By your train of though, should we not split 
out the "everything else" bucket into a different callback method for 
every different error? If no, why not, but only for serialization errors?

 From what I believe to remember, historically, we added the 
ProductionExceptionHandler, and kinda just missed the serialization 
error case. And later, when we extended the handler we just could not 
re-use the existing callback as it was typed with `<byte[], byte[]>` and 
it would have been an incompatible change; so it was rather a workaround 
to add the second method to then handler, but not really intended design?


It's of course only my personal opinion that I believe a single callback 
method is simpler/cleaner compared to sticking with two, and adding the 
new exception type to make it backward compatible seems worth it. It 
also kinda introduces the same patter we use elsewhere (cf KIP-1036) 
what I actually think is an argument for introducing 
`RercordSerializationExcetpion`, to unify user experience across the board.

Would be great to hear from others about this point. It's not that I 
strongly object to having two methods, and I would not block this KIP on 
this question.



-Matthias


On 5/7/24 3:40 PM, Sophie Blee-Goldman wrote:
> First of all, I agree that it makes sense to maintain the two separate
> callbacks for the ProductionExceptionHandler, since one of them is
> specifically for serialization exceptions while the other is used for
> everything/anything else. I also think we can take advantage of this fact
> to simplify things a bit and cut down on the amount of new stuff added to
> the API by just adding a parameter to the #handleSerializationException
> callback and use that to pass in the SerializationExceptionOrigin enum to
> distinguish between key vs value. This way we wouldn't need to introduce
> yet another exception type (the RecordSerializationException) just to pass
> in this information.
> 
> Thoughts?
> 
> On Tue, May 7, 2024 at 8:33 AM Loic Greffier <lo...@michelin.com>
> wrote:
> 
>> Hi Matthias,
>>
>> To sum up with the ProductionExceptionHandler callback methods (106)
>> proposed changes.
>>
>> A new method ProductionExceptionHandler#handle is added with the following
>> signature:
>>
>>> ProductionExceptionHandlerResponse handle(final ErrorHandlerContext
>> context, final ProducerRecord<?,?> record, final Exception exception);
>>
>> The ProducerRecord<?,?> parameter has changed to accept a serialized or
>> non-serialized record.
>> Thus, the new ProductionExceptionHandler#handle method can handle both
>> production exception and serialization exception.
>>
>> Both old ProductionExceptionHandler#handle and
>> ProductionExceptionHandler#handleSerializationException methods are now
>> deprecated.
>> The old ProductionExceptionHandler#handle method gets a default
>> implementation, so users do not have to implement a deprecated method.
>>
>> To handle backward compatibility, the new
>> ProductionExceptionHandler#handle method gets a default implementation.
>>
>>> default ProductionExceptionHandlerResponse handle(final
>> ErrorHandlerContext context, final ProducerRecord<?, ?> record, final
>> Exception exception) {
>>>    if (exception instanceof RecordSerializationException) {
>>>        this.handleSerializationException(record, exception.getCause());
>>>    }
>>>
>>>    return handle((ProducerRecord<byte[], byte[]>) record, exception);
>>> }
>>
>> The default implementation either invokes #handleSerializationException or
>> #handle depending on the type of the exception, thus users still relying on
>> deprecated ProductionExceptionHandler#handle
>> or ProductionExceptionHandler#handleSerializationException custom
>> implementations won't break.
>>
>> The new ProductionExceptionHandler#handle method is now invoked in case of
>> serialization exception:
>>
>>> public <K, V> void send(final String topic, final K key, final V value,
>> ...) {
>>>      try {
>>>          keyBytes = keySerializer.serialize(topic, headers, key);
>>>          ...
>>>      } catch (final ClassCastException exception) {
>>>        ...
>>>      } catch (final Exception exception) {
>>>
>>>          try {
>>>              response = productionExceptionHandler.handle(context,
>> record, new RecordSerializationException(SerializationExceptionOrigin.KEY,
>> exception));
>>>          } catch (final Exception e) {
>>>          ...
>>>          }
>>>      }
>>> }
>>
>> To wrap the origin serialization exception and determine whether it comes
>> from the key or the value, a new exception class is created:
>>
>>> public class RecordSerializationException extends SerializationException
>> {
>>>      public enum SerializationExceptionOrigin {
>>>          KEY,
>>>          VALUE
>>>      }
>>>
>>>      public RecordSerializationException(SerializationExceptionOrigin
>> origin, final Throwable cause);
>>> }
>>
>> Hope it all makes sense,
>> Loïc
>>
> 

Re: [DISCUSS] KIP-1033: Add Kafka Streams exception handler for exceptions occuring during processing

[Sophie Blee-Goldman <so...@responsive.dev>]

First of all, I agree that it makes sense to maintain the two separate
callbacks for the ProductionExceptionHandler, since one of them is
specifically for serialization exceptions while the other is used for
everything/anything else. I also think we can take advantage of this fact
to simplify things a bit and cut down on the amount of new stuff added to
the API by just adding a parameter to the #handleSerializationException
callback and use that to pass in the SerializationExceptionOrigin enum to
distinguish between key vs value. This way we wouldn't need to introduce
yet another exception type (the RecordSerializationException) just to pass
in this information.

Thoughts?

On Tue, May 7, 2024 at 8:33 AM Loic Greffier <lo...@michelin.com>
wrote:

> Hi Matthias,
>
> To sum up with the ProductionExceptionHandler callback methods (106)
> proposed changes.
>
> A new method ProductionExceptionHandler#handle is added with the following
> signature:
>
> > ProductionExceptionHandlerResponse handle(final ErrorHandlerContext
> context, final ProducerRecord<?,?> record, final Exception exception);
>
> The ProducerRecord<?,?> parameter has changed to accept a serialized or
> non-serialized record.
> Thus, the new ProductionExceptionHandler#handle method can handle both
> production exception and serialization exception.
>
> Both old ProductionExceptionHandler#handle and
> ProductionExceptionHandler#handleSerializationException methods are now
> deprecated.
> The old ProductionExceptionHandler#handle method gets a default
> implementation, so users do not have to implement a deprecated method.
>
> To handle backward compatibility, the new
> ProductionExceptionHandler#handle method gets a default implementation.
>
> > default ProductionExceptionHandlerResponse handle(final
> ErrorHandlerContext context, final ProducerRecord<?, ?> record, final
> Exception exception) {
> >   if (exception instanceof RecordSerializationException) {
> >       this.handleSerializationException(record, exception.getCause());
> >   }
> >
> >   return handle((ProducerRecord<byte[], byte[]>) record, exception);
> > }
>
> The default implementation either invokes #handleSerializationException or
> #handle depending on the type of the exception, thus users still relying on
> deprecated ProductionExceptionHandler#handle
> or ProductionExceptionHandler#handleSerializationException custom
> implementations won't break.
>
> The new ProductionExceptionHandler#handle method is now invoked in case of
> serialization exception:
>
> > public <K, V> void send(final String topic, final K key, final V value,
> ...) {
> >     try {
> >         keyBytes = keySerializer.serialize(topic, headers, key);
> >         ...
> >     } catch (final ClassCastException exception) {
> >       ...
> >     } catch (final Exception exception) {
> >
> >         try {
> >             response = productionExceptionHandler.handle(context,
> record, new RecordSerializationException(SerializationExceptionOrigin.KEY,
> exception));
> >         } catch (final Exception e) {
> >         ...
> >         }
> >     }
> > }
>
> To wrap the origin serialization exception and determine whether it comes
> from the key or the value, a new exception class is created:
>
> > public class RecordSerializationException extends SerializationException
> {
> >     public enum SerializationExceptionOrigin {
> >         KEY,
> >         VALUE
> >     }
> >
> >     public RecordSerializationException(SerializationExceptionOrigin
> origin, final Throwable cause);
> > }
>
> Hope it all makes sense,
> Loïc
>

[DISCUSS] KIP-1033: Add Kafka Streams exception handler for exceptions occuring during processing

[Loic Greffier <lo...@michelin.com>]

Hi Matthias, 

To sum up with the ProductionExceptionHandler callback methods (106) proposed changes.

A new method ProductionExceptionHandler#handle is added with the following signature:

> ProductionExceptionHandlerResponse handle(final ErrorHandlerContext context, final ProducerRecord<?,?> record, final Exception exception);

The ProducerRecord<?,?> parameter has changed to accept a serialized or non-serialized record. 
Thus, the new ProductionExceptionHandler#handle method can handle both production exception and serialization exception.

Both old ProductionExceptionHandler#handle and ProductionExceptionHandler#handleSerializationException methods are now deprecated.
The old ProductionExceptionHandler#handle method gets a default implementation, so users do not have to implement a deprecated method.

To handle backward compatibility, the new ProductionExceptionHandler#handle method gets a default implementation.

> default ProductionExceptionHandlerResponse handle(final ErrorHandlerContext context, final ProducerRecord<?, ?> record, final Exception exception) {
>   if (exception instanceof RecordSerializationException) {
>       this.handleSerializationException(record, exception.getCause());
>   }
>
>   return handle((ProducerRecord<byte[], byte[]>) record, exception);
> }

The default implementation either invokes #handleSerializationException or #handle depending on the type of the exception, thus users still relying on deprecated ProductionExceptionHandler#handle 
or ProductionExceptionHandler#handleSerializationException custom implementations won't break.

The new ProductionExceptionHandler#handle method is now invoked in case of serialization exception:

> public <K, V> void send(final String topic, final K key, final V value, ...) {
>     try {
>         keyBytes = keySerializer.serialize(topic, headers, key);
>         ...	
>     } catch (final ClassCastException exception) {
>       ...
>     } catch (final Exception exception) {
>     
>         try {
>             response = productionExceptionHandler.handle(context, record, new RecordSerializationException(SerializationExceptionOrigin.KEY, exception));
>         } catch (final Exception e) {
>         ...
>         }
>     }
> }

To wrap the origin serialization exception and determine whether it comes from the key or the value, a new exception class is created:

> public class RecordSerializationException extends SerializationException {
>     public enum SerializationExceptionOrigin {
>         KEY,
>         VALUE
>     }
>
>     public RecordSerializationException(SerializationExceptionOrigin origin, final Throwable cause);
> }

Hope it all makes sense,
Loïc

Re: [DISCUSS] KIP-1033: Add Kafka Streams exception handler for exceptions occuring during processing

["Matthias J. Sax" <mj...@apache.org>]

What about (106) to unify both exiting callback methods of 
`ProductionExceptionHandler` into a single one, instead of adding two 
new ones?

Damien's last reply about it was:

> I will think about unifying, I do agree it would be cleaner.

There was not follow up on this question, and the KIP right now still 
proposes to add two new methods, which I believe we could (should?) 
unify to:

> default ProductionExceptionHandlerResponse handle(final ErrorHandlerContext context, 
>                                                   final ProducerRecord<?, ?> record,
>                                                   final Exception exception) {

Ie, we drop the generics `<byte[], byte[]>` on `ProducerRecord` what 
allows you to also pass in a non-serialized ProducerRecord of any type 
for the serialization error case.

Btw: wondering if we also want to pass in a flag/enum about key vs value 
serialization error similar to what was proposed in KIP-1036? The only 
"oddity" would be, that we call the handler other error cases, too, not 
just for serialization exceptions. But we wculd tackle this by 
introducing a new class `RecordSerializationException` which would 
include the flag and would ensure that KS hands this exception into the 
handler. This would keep the handler interface/method itself clean.


-Matthias




On 5/3/24 2:15 AM, Loic Greffier wrote:
> Hi Bruno,
> 
> Good catch, KIP has been updated accordingly.
> 
> Regards,
> Loïc

Re: [DISCUSS] KIP-1033: Add Kafka Streams exception handler for exceptions occuring during processing

[Loic Greffier <lo...@michelin.com>]

Hi Bruno, 

Good catch, KIP has been updated accordingly.

Regards,
Loïc

Re: [DISCUSS] KIP-1033: Add Kafka Streams exception handler for exceptions occuring during processing

[Bruno Cadonna <ca...@apache.org>]

Hi,

the KIP looks great!

public static final String PROCESS_EXCEPTION_HANDLER_CLASS_CONFIG = 
"process.exception.handler".

needs to be changed to

public static final String PROCESSING_EXCEPTION_HANDLER_CLASS_CONFIG = 
"processing.exception.handler".

The name of the constant has been already corrected in the code block 
but the actual name of the config (i.e., the content of the constant) 
has not been changed yet.


Best,
Bruno


On 5/3/24 10:35 AM, Sebastien Viale wrote:
> Hi,
> So, we all agree to revert to the regular Headers interface in ErrorHandlerContext.
> We will update the KIP accordingly.
> @Sophie => Yes, this is the last remaining question, and it has been open for voting since last week.
> Thanks
> 
> Sébastien
> ________________________________
> De : Andrew Schofield <an...@live.com>
> Envoyé : vendredi 3 mai 2024 06:44
> À : dev@kafka.apache.org <de...@kafka.apache.org>
> Objet : [SUSPICIOUS EXTERNAL MESSAGE] Re: [DISCUSS] KIP-1033: Add Kafka Streams exception handler for exceptions occuring during processing
> 
> Warning This might be a fraudulent message! When clicking REPLY, your answers will NOT go to the sender (andrew_schofield@live.com). Instead, replies will be sent to dev@kafka.apache.org. Be cautious!
> 
> Hi,
> I’ve changed my mind on this one having read through the comments.
> 
> I don’t think the exception handler should be able to mess with the headers
> to the detriment of the code that called the handler.
> 
> While I like the hygiene of having an ImmutableHeaders interface,
> I feel we can use the existing interface to get the effect we desire.
> 
> Thanks,
> Andrew
> 
> ========================================================================================
> This email was screened for spam and malicious content but exercise caution anyway.
> 
> 
>> On 3 May 2024, at 03:40, Sophie Blee-Goldman <so...@responsive.dev> wrote:
>>
>> I tend to agree that we should just return a pure Headers instead of
>> introducing a new class/interface to protect overwriting them. I think a
>> pretty good case has been made already so I won't add onto it, just wanted
>> to voice my support.
>>
>> Is that the only remaining question on this KIP? Might be ok to move to a
>> vote now?
>>
>> On Wed, May 1, 2024 at 8:05 AM Lianet M. <li...@gmail.com> wrote:
>>
>>> Hi all, thanks Damien for the KIP!
>>>
>>> After looking into the KIP and comments, my only concern is aligned with
>>> one of Matthias comments, around the ImmutableHeaders introduction, with
>>> the motivation not being clear enough. The existing handlers already expose
>>> the headers (indirectly). Ex.
>>> ProductionExceptionHandler.handleSerializationException provides the
>>> ProducerRecord as an argument, so they are already exposed in those
>>> callbacks through record.headers(). Is there a reason to think that it
>>> would be a problem to expose the headers in the
>>> new ProcessingExceptionHandler, but that it's not a problem for the
>>> existing handler?
>>>
>>> If there is no real concern about the KS engine requiring those headers, it
>>> feels hard to mentally justify the complexity we transfer to the user by
>>> exposing a new concept into the callbacks to represent the headers. In the
>>> end, it strays aways from the simple/consistent representation of Headers
>>> used all over. Even if eventually the KS engine needs to use the headers
>>> after the callbacks with certainty that they were not altered, still feels
>>> like it's something we could attempt to solve internally, without having to
>>> transfer "new concepts" into the user (ex. the deep-copy as it was
>>> suggested, seems like the kind of trade-off that would maybe be acceptable
>>> here to gain simplicity and consistency among the handlers with a single
>>> existing representation of Headers).
>>>
>>> Best!
>>>
>>> Lianet
>>>
>>>
>>>
>>> On Tue, Apr 30, 2024 at 9:36 PM Matthias J. Sax <mj...@apache.org> wrote:
>>>
>>>> Thanks for the update.
>>>>
>>>> I am wondering if we should use `ReadOnlyHeaders` instead of
>>>> `ImmutableHeaders` as interface name?
>>>>
>>>> Also, the returned `Header` interface is technically not immutable
>>>> either, because `Header#key()` returns a mutable byte-array... Would we
>>>> need a `ReadOnlyHeader` interface?
>>>>
>>>> If yes, it seems that `ReadOnlyHeaders` should not be a super-interface
>>>> of `Headers` but it would rather be a standalone interface, and a
>>>> wrapper for a `Headers` instance? And `ReadOnlyHeader` would return some
>>>> immutable type instead of `byte[]` for the value()?
>>>>
>>>> An alternative would be to deep-copy the value byte-array what would not
>>>> be free, but given that we are talking about exception handling, it
>>>> would not be on the hot code path, and thus might be acceptable?
>>>>
>>>>
>>>> The above seems to increase the complexity significantly though. Hence,
>>>> I have seconds thoughts on the immutability question:
>>>>
>>>> Do we really need to worry about mutability after all, because in the
>>>> end, KS runtime won't read the Headers instance after the handler was
>>>> called, and if a user modifies the passed in headers, there won't be any
>>>> actual damage (ie, no side effects)? For this case, it might even be ok
>>>> to also not add `ImmutableHeaders` to begin with?
>>>>
>>>>
>>>>
>>>> Sorry for the forth and back (yes, forth and back, because back and
>>>> forth does not make sense -- it's not logical -- just trying to fix
>>>> English :D) as I did bring up the immutability question in the first
>>>> place...
>>>>
>>>>
>>>>
>>>> -Matthias
>>>>
>>>> On 4/25/24 5:56 AM, Loic Greffier wrote:
>>>>> Hi Matthias,
>>>>>
>>>>> I have updated the KIP regarding points 103 and 108.
>>>>>
>>>>> 103.
>>>>> I have suggested a new `ImmutableHeaders` interface to deal with the
>>>>> immutability concern of the headers, which is basically the `Headers`
>>>>> interface without the write accesses.
>>>>>
>>>>> public interface ImmutableHeaders {
>>>>>      Header lastHeader(String key);
>>>>>      Iterable<Header> headers(String key);
>>>>>      Header[] toArray();
>>>>> }
>>>>>
>>>>> The `Headers` interface can be updated accordingly:
>>>>>
>>>>> public interface Headers extends ImmutableHeaders, Iterable<Header> {
>>>>>      //…
>>>>> }
>>>>>
>>>>> Loïc
>>>>
>>>

[DISCUSS] KIP-1033: Add Kafka Streams exception handler for exceptions occuring during processing

[Sebastien Viale <se...@michelin.com>]

Just as a reminder, here the link for the vote:

https://lists.apache.org/thread/o1kvv8zjzsp72ohcjpckdy544ko9tjjb


regards
Sébastien


________________________________
De : Sebastien Viale <se...@michelin.com>
Envoyé : vendredi 3 mai 2024 10:35
À : dev@kafka.apache.org <de...@kafka.apache.org>
Objet : [DISCUSS] KIP-1033: Add Kafka Streams exception handler for exceptions occuring during processing

Hi,
So, we all agree to revert to the regular Headers interface in ErrorHandlerContext.
We will update the KIP accordingly.
@Sophie => Yes, this is the last remaining question, and it has been open for voting since last week.
Thanks

Sébastien
________________________________
De : Andrew Schofield <an...@live.com>
Envoyé : vendredi 3 mai 2024 06:44
À : dev@kafka.apache.org <de...@kafka.apache.org>
Objet : [SUSPICIOUS EXTERNAL MESSAGE] Re: [DISCUSS] KIP-1033: Add Kafka Streams exception handler for exceptions occuring during processing

Warning This might be a fraudulent message! When clicking REPLY, your answers will NOT go to the sender (andrew_schofield@live.com). Instead, replies will be sent to dev@kafka.apache.org. Be cautious!

Hi,
I’ve changed my mind on this one having read through the comments.

I don’t think the exception handler should be able to mess with the headers
to the detriment of the code that called the handler.

While I like the hygiene of having an ImmutableHeaders interface,
I feel we can use the existing interface to get the effect we desire.

Thanks,
Andrew

========================================================================================
This email was screened for spam and malicious content but exercise caution anyway.


> On 3 May 2024, at 03:40, Sophie Blee-Goldman <so...@responsive.dev> wrote:
>
> I tend to agree that we should just return a pure Headers instead of
> introducing a new class/interface to protect overwriting them. I think a
> pretty good case has been made already so I won't add onto it, just wanted
> to voice my support.
>
> Is that the only remaining question on this KIP? Might be ok to move to a
> vote now?
>
> On Wed, May 1, 2024 at 8:05 AM Lianet M. <li...@gmail.com> wrote:
>
>> Hi all, thanks Damien for the KIP!
>>
>> After looking into the KIP and comments, my only concern is aligned with
>> one of Matthias comments, around the ImmutableHeaders introduction, with
>> the motivation not being clear enough. The existing handlers already expose
>> the headers (indirectly). Ex.
>> ProductionExceptionHandler.handleSerializationException provides the
>> ProducerRecord as an argument, so they are already exposed in those
>> callbacks through record.headers(). Is there a reason to think that it
>> would be a problem to expose the headers in the
>> new ProcessingExceptionHandler, but that it's not a problem for the
>> existing handler?
>>
>> If there is no real concern about the KS engine requiring those headers, it
>> feels hard to mentally justify the complexity we transfer to the user by
>> exposing a new concept into the callbacks to represent the headers. In the
>> end, it strays aways from the simple/consistent representation of Headers
>> used all over. Even if eventually the KS engine needs to use the headers
>> after the callbacks with certainty that they were not altered, still feels
>> like it's something we could attempt to solve internally, without having to
>> transfer "new concepts" into the user (ex. the deep-copy as it was
>> suggested, seems like the kind of trade-off that would maybe be acceptable
>> here to gain simplicity and consistency among the handlers with a single
>> existing representation of Headers).
>>
>> Best!
>>
>> Lianet
>>
>>
>>
>> On Tue, Apr 30, 2024 at 9:36 PM Matthias J. Sax <mj...@apache.org> wrote:
>>
>>> Thanks for the update.
>>>
>>> I am wondering if we should use `ReadOnlyHeaders` instead of
>>> `ImmutableHeaders` as interface name?
>>>
>>> Also, the returned `Header` interface is technically not immutable
>>> either, because `Header#key()` returns a mutable byte-array... Would we
>>> need a `ReadOnlyHeader` interface?
>>>
>>> If yes, it seems that `ReadOnlyHeaders` should not be a super-interface
>>> of `Headers` but it would rather be a standalone interface, and a
>>> wrapper for a `Headers` instance? And `ReadOnlyHeader` would return some
>>> immutable type instead of `byte[]` for the value()?
>>>
>>> An alternative would be to deep-copy the value byte-array what would not
>>> be free, but given that we are talking about exception handling, it
>>> would not be on the hot code path, and thus might be acceptable?
>>>
>>>
>>> The above seems to increase the complexity significantly though. Hence,
>>> I have seconds thoughts on the immutability question:
>>>
>>> Do we really need to worry about mutability after all, because in the
>>> end, KS runtime won't read the Headers instance after the handler was
>>> called, and if a user modifies the passed in headers, there won't be any
>>> actual damage (ie, no side effects)? For this case, it might even be ok
>>> to also not add `ImmutableHeaders` to begin with?
>>>
>>>
>>>
>>> Sorry for the forth and back (yes, forth and back, because back and
>>> forth does not make sense -- it's not logical -- just trying to fix
>>> English :D) as I did bring up the immutability question in the first
>>> place...
>>>
>>>
>>>
>>> -Matthias
>>>
>>> On 4/25/24 5:56 AM, Loic Greffier wrote:
>>>> Hi Matthias,
>>>>
>>>> I have updated the KIP regarding points 103 and 108.
>>>>
>>>> 103.
>>>> I have suggested a new `ImmutableHeaders` interface to deal with the
>>>> immutability concern of the headers, which is basically the `Headers`
>>>> interface without the write accesses.
>>>>
>>>> public interface ImmutableHeaders {
>>>>     Header lastHeader(String key);
>>>>     Iterable<Header> headers(String key);
>>>>     Header[] toArray();
>>>> }
>>>>
>>>> The `Headers` interface can be updated accordingly:
>>>>
>>>> public interface Headers extends ImmutableHeaders, Iterable<Header> {
>>>>     //…
>>>> }
>>>>
>>>> Loïc
>>>
>>

[DISCUSS] KIP-1033: Add Kafka Streams exception handler for exceptions occuring during processing

[Sebastien Viale <se...@michelin.com>]

Hi,
So, we all agree to revert to the regular Headers interface in ErrorHandlerContext.
We will update the KIP accordingly.
@Sophie => Yes, this is the last remaining question, and it has been open for voting since last week.
Thanks

Sébastien
________________________________
De : Andrew Schofield <an...@live.com>
Envoyé : vendredi 3 mai 2024 06:44
À : dev@kafka.apache.org <de...@kafka.apache.org>
Objet : [SUSPICIOUS EXTERNAL MESSAGE] Re: [DISCUSS] KIP-1033: Add Kafka Streams exception handler for exceptions occuring during processing

Warning This might be a fraudulent message! When clicking REPLY, your answers will NOT go to the sender (andrew_schofield@live.com). Instead, replies will be sent to dev@kafka.apache.org. Be cautious!

Hi,
I’ve changed my mind on this one having read through the comments.

I don’t think the exception handler should be able to mess with the headers
to the detriment of the code that called the handler.

While I like the hygiene of having an ImmutableHeaders interface,
I feel we can use the existing interface to get the effect we desire.

Thanks,
Andrew

========================================================================================
This email was screened for spam and malicious content but exercise caution anyway.


> On 3 May 2024, at 03:40, Sophie Blee-Goldman <so...@responsive.dev> wrote:
>
> I tend to agree that we should just return a pure Headers instead of
> introducing a new class/interface to protect overwriting them. I think a
> pretty good case has been made already so I won't add onto it, just wanted
> to voice my support.
>
> Is that the only remaining question on this KIP? Might be ok to move to a
> vote now?
>
> On Wed, May 1, 2024 at 8:05 AM Lianet M. <li...@gmail.com> wrote:
>
>> Hi all, thanks Damien for the KIP!
>>
>> After looking into the KIP and comments, my only concern is aligned with
>> one of Matthias comments, around the ImmutableHeaders introduction, with
>> the motivation not being clear enough. The existing handlers already expose
>> the headers (indirectly). Ex.
>> ProductionExceptionHandler.handleSerializationException provides the
>> ProducerRecord as an argument, so they are already exposed in those
>> callbacks through record.headers(). Is there a reason to think that it
>> would be a problem to expose the headers in the
>> new ProcessingExceptionHandler, but that it's not a problem for the
>> existing handler?
>>
>> If there is no real concern about the KS engine requiring those headers, it
>> feels hard to mentally justify the complexity we transfer to the user by
>> exposing a new concept into the callbacks to represent the headers. In the
>> end, it strays aways from the simple/consistent representation of Headers
>> used all over. Even if eventually the KS engine needs to use the headers
>> after the callbacks with certainty that they were not altered, still feels
>> like it's something we could attempt to solve internally, without having to
>> transfer "new concepts" into the user (ex. the deep-copy as it was
>> suggested, seems like the kind of trade-off that would maybe be acceptable
>> here to gain simplicity and consistency among the handlers with a single
>> existing representation of Headers).
>>
>> Best!
>>
>> Lianet
>>
>>
>>
>> On Tue, Apr 30, 2024 at 9:36 PM Matthias J. Sax <mj...@apache.org> wrote:
>>
>>> Thanks for the update.
>>>
>>> I am wondering if we should use `ReadOnlyHeaders` instead of
>>> `ImmutableHeaders` as interface name?
>>>
>>> Also, the returned `Header` interface is technically not immutable
>>> either, because `Header#key()` returns a mutable byte-array... Would we
>>> need a `ReadOnlyHeader` interface?
>>>
>>> If yes, it seems that `ReadOnlyHeaders` should not be a super-interface
>>> of `Headers` but it would rather be a standalone interface, and a
>>> wrapper for a `Headers` instance? And `ReadOnlyHeader` would return some
>>> immutable type instead of `byte[]` for the value()?
>>>
>>> An alternative would be to deep-copy the value byte-array what would not
>>> be free, but given that we are talking about exception handling, it
>>> would not be on the hot code path, and thus might be acceptable?
>>>
>>>
>>> The above seems to increase the complexity significantly though. Hence,
>>> I have seconds thoughts on the immutability question:
>>>
>>> Do we really need to worry about mutability after all, because in the
>>> end, KS runtime won't read the Headers instance after the handler was
>>> called, and if a user modifies the passed in headers, there won't be any
>>> actual damage (ie, no side effects)? For this case, it might even be ok
>>> to also not add `ImmutableHeaders` to begin with?
>>>
>>>
>>>
>>> Sorry for the forth and back (yes, forth and back, because back and
>>> forth does not make sense -- it's not logical -- just trying to fix
>>> English :D) as I did bring up the immutability question in the first
>>> place...
>>>
>>>
>>>
>>> -Matthias
>>>
>>> On 4/25/24 5:56 AM, Loic Greffier wrote:
>>>> Hi Matthias,
>>>>
>>>> I have updated the KIP regarding points 103 and 108.
>>>>
>>>> 103.
>>>> I have suggested a new `ImmutableHeaders` interface to deal with the
>>>> immutability concern of the headers, which is basically the `Headers`
>>>> interface without the write accesses.
>>>>
>>>> public interface ImmutableHeaders {
>>>>     Header lastHeader(String key);
>>>>     Iterable<Header> headers(String key);
>>>>     Header[] toArray();
>>>> }
>>>>
>>>> The `Headers` interface can be updated accordingly:
>>>>
>>>> public interface Headers extends ImmutableHeaders, Iterable<Header> {
>>>>     //…
>>>> }
>>>>
>>>> Loïc
>>>
>>

Re: [DISCUSS] KIP-1033: Add Kafka Streams exception handler for exceptions occuring during processing

[Andrew Schofield <an...@live.com>]

Hi,
I’ve changed my mind on this one having read through the comments.

I don’t think the exception handler should be able to mess with the headers
to the detriment of the code that called the handler.

While I like the hygiene of having an ImmutableHeaders interface,
I feel we can use the existing interface to get the effect we desire.

Thanks,
Andrew

> On 3 May 2024, at 03:40, Sophie Blee-Goldman <so...@responsive.dev> wrote:
> 
> I tend to agree that we should just return a pure Headers instead of
> introducing a new class/interface to protect overwriting them. I think a
> pretty good case has been made already so I won't add onto it, just wanted
> to voice my support.
> 
> Is that the only remaining question on this KIP? Might be ok to move to a
> vote now?
> 
> On Wed, May 1, 2024 at 8:05 AM Lianet M. <li...@gmail.com> wrote:
> 
>> Hi all, thanks Damien for the KIP!
>> 
>> After looking into the KIP and comments, my only concern is aligned with
>> one of Matthias comments, around the ImmutableHeaders introduction, with
>> the motivation not being clear enough. The existing handlers already expose
>> the headers (indirectly). Ex.
>> ProductionExceptionHandler.handleSerializationException provides the
>> ProducerRecord as an argument, so they are already exposed in those
>> callbacks through record.headers(). Is there a reason to think that it
>> would be a problem to expose the headers in the
>> new ProcessingExceptionHandler, but that it's not a problem for the
>> existing handler?
>> 
>> If there is no real concern about the KS engine requiring those headers, it
>> feels hard to mentally justify the complexity we transfer to the user by
>> exposing a new concept into the callbacks to represent the headers. In the
>> end, it strays aways from the simple/consistent representation of Headers
>> used all over. Even if eventually the KS engine needs to use the headers
>> after the callbacks with certainty that they were not altered, still feels
>> like it's something we could attempt to solve internally, without having to
>> transfer "new concepts" into the user (ex. the deep-copy as it was
>> suggested, seems like the kind of trade-off that would maybe be acceptable
>> here to gain simplicity and consistency among the handlers with a single
>> existing representation of Headers).
>> 
>> Best!
>> 
>> Lianet
>> 
>> 
>> 
>> On Tue, Apr 30, 2024 at 9:36 PM Matthias J. Sax <mj...@apache.org> wrote:
>> 
>>> Thanks for the update.
>>> 
>>> I am wondering if we should use `ReadOnlyHeaders` instead of
>>> `ImmutableHeaders` as interface name?
>>> 
>>> Also, the returned `Header` interface is technically not immutable
>>> either, because `Header#key()` returns a mutable byte-array... Would we
>>> need a `ReadOnlyHeader` interface?
>>> 
>>> If yes, it seems that `ReadOnlyHeaders` should not be a super-interface
>>> of `Headers` but it would rather be a standalone interface, and a
>>> wrapper for a `Headers` instance? And `ReadOnlyHeader` would return some
>>> immutable type instead of `byte[]` for the value()?
>>> 
>>> An alternative would be to deep-copy the value byte-array what would not
>>> be free, but given that we are talking about exception handling, it
>>> would not be on the hot code path, and thus might be acceptable?
>>> 
>>> 
>>> The above seems to increase the complexity significantly though. Hence,
>>> I have seconds thoughts on the immutability question:
>>> 
>>> Do we really need to worry about mutability after all, because in the
>>> end, KS runtime won't read the Headers instance after the handler was
>>> called, and if a user modifies the passed in headers, there won't be any
>>> actual damage (ie, no side effects)? For this case, it might even be ok
>>> to also not add `ImmutableHeaders` to begin with?
>>> 
>>> 
>>> 
>>> Sorry for the forth and back (yes, forth and back, because back and
>>> forth does not make sense -- it's not logical -- just trying to fix
>>> English :D) as I did bring up the immutability question in the first
>>> place...
>>> 
>>> 
>>> 
>>> -Matthias
>>> 
>>> On 4/25/24 5:56 AM, Loic Greffier wrote:
>>>> Hi Matthias,
>>>> 
>>>> I have updated the KIP regarding points 103 and 108.
>>>> 
>>>> 103.
>>>> I have suggested a new `ImmutableHeaders` interface to deal with the
>>>> immutability concern of the headers, which is basically the `Headers`
>>>> interface without the write accesses.
>>>> 
>>>> public interface ImmutableHeaders {
>>>>     Header lastHeader(String key);
>>>>     Iterable<Header> headers(String key);
>>>>     Header[] toArray();
>>>> }
>>>> 
>>>> The `Headers` interface can be updated accordingly:
>>>> 
>>>> public interface Headers extends ImmutableHeaders, Iterable<Header> {
>>>>     //…
>>>> }
>>>> 
>>>> Loïc
>>> 
>> 

Re: [DISCUSS] KIP-1033: Add Kafka Streams exception handler for exceptions occuring during processing

[Sophie Blee-Goldman <so...@responsive.dev>]

I tend to agree that we should just return a pure Headers instead of
introducing a new class/interface to protect overwriting them. I think a
pretty good case has been made already so I won't add onto it, just wanted
to voice my support.

Is that the only remaining question on this KIP? Might be ok to move to a
vote now?

On Wed, May 1, 2024 at 8:05 AM Lianet M. <li...@gmail.com> wrote:

> Hi all, thanks Damien for the KIP!
>
> After looking into the KIP and comments, my only concern is aligned with
> one of Matthias comments, around the ImmutableHeaders introduction, with
> the motivation not being clear enough. The existing handlers already expose
> the headers (indirectly). Ex.
> ProductionExceptionHandler.handleSerializationException provides the
> ProducerRecord as an argument, so they are already exposed in those
> callbacks through record.headers(). Is there a reason to think that it
> would be a problem to expose the headers in the
> new ProcessingExceptionHandler, but that it's not a problem for the
> existing handler?
>
> If there is no real concern about the KS engine requiring those headers, it
> feels hard to mentally justify the complexity we transfer to the user by
> exposing a new concept into the callbacks to represent the headers. In the
> end, it strays aways from the simple/consistent representation of Headers
> used all over. Even if eventually the KS engine needs to use the headers
> after the callbacks with certainty that they were not altered, still feels
> like it's something we could attempt to solve internally, without having to
> transfer "new concepts" into the user (ex. the deep-copy as it was
> suggested, seems like the kind of trade-off that would maybe be acceptable
> here to gain simplicity and consistency among the handlers with a single
> existing representation of Headers).
>
> Best!
>
> Lianet
>
>
>
> On Tue, Apr 30, 2024 at 9:36 PM Matthias J. Sax <mj...@apache.org> wrote:
>
> > Thanks for the update.
> >
> > I am wondering if we should use `ReadOnlyHeaders` instead of
> > `ImmutableHeaders` as interface name?
> >
> > Also, the returned `Header` interface is technically not immutable
> > either, because `Header#key()` returns a mutable byte-array... Would we
> > need a `ReadOnlyHeader` interface?
> >
> > If yes, it seems that `ReadOnlyHeaders` should not be a super-interface
> > of `Headers` but it would rather be a standalone interface, and a
> > wrapper for a `Headers` instance? And `ReadOnlyHeader` would return some
> > immutable type instead of `byte[]` for the value()?
> >
> > An alternative would be to deep-copy the value byte-array what would not
> > be free, but given that we are talking about exception handling, it
> > would not be on the hot code path, and thus might be acceptable?
> >
> >
> > The above seems to increase the complexity significantly though. Hence,
> > I have seconds thoughts on the immutability question:
> >
> > Do we really need to worry about mutability after all, because in the
> > end, KS runtime won't read the Headers instance after the handler was
> > called, and if a user modifies the passed in headers, there won't be any
> > actual damage (ie, no side effects)? For this case, it might even be ok
> > to also not add `ImmutableHeaders` to begin with?
> >
> >
> >
> > Sorry for the forth and back (yes, forth and back, because back and
> > forth does not make sense -- it's not logical -- just trying to fix
> > English :D) as I did bring up the immutability question in the first
> > place...
> >
> >
> >
> > -Matthias
> >
> > On 4/25/24 5:56 AM, Loic Greffier wrote:
> > > Hi Matthias,
> > >
> > > I have updated the KIP regarding points 103 and 108.
> > >
> > > 103.
> > > I have suggested a new `ImmutableHeaders` interface to deal with the
> > > immutability concern of the headers, which is basically the `Headers`
> > > interface without the write accesses.
> > >
> > > public interface ImmutableHeaders {
> > >      Header lastHeader(String key);
> > >      Iterable<Header> headers(String key);
> > >      Header[] toArray();
> > > }
> > >
> > > The `Headers` interface can be updated accordingly:
> > >
> > > public interface Headers extends ImmutableHeaders, Iterable<Header> {
> > >      //…
> > > }
> > >
> > > Loïc
> >
>

Re: [DISCUSS] KIP-1033: Add Kafka Streams exception handler for exceptions occuring during processing

["Lianet M." <li...@gmail.com>]

Hi all, thanks Damien for the KIP!

After looking into the KIP and comments, my only concern is aligned with
one of Matthias comments, around the ImmutableHeaders introduction, with
the motivation not being clear enough. The existing handlers already expose
the headers (indirectly). Ex.
ProductionExceptionHandler.handleSerializationException provides the
ProducerRecord as an argument, so they are already exposed in those
callbacks through record.headers(). Is there a reason to think that it
would be a problem to expose the headers in the
new ProcessingExceptionHandler, but that it's not a problem for the
existing handler?

If there is no real concern about the KS engine requiring those headers, it
feels hard to mentally justify the complexity we transfer to the user by
exposing a new concept into the callbacks to represent the headers. In the
end, it strays aways from the simple/consistent representation of Headers
used all over. Even if eventually the KS engine needs to use the headers
after the callbacks with certainty that they were not altered, still feels
like it's something we could attempt to solve internally, without having to
transfer "new concepts" into the user (ex. the deep-copy as it was
suggested, seems like the kind of trade-off that would maybe be acceptable
here to gain simplicity and consistency among the handlers with a single
existing representation of Headers).

Best!

Lianet



On Tue, Apr 30, 2024 at 9:36 PM Matthias J. Sax <mj...@apache.org> wrote:

> Thanks for the update.
>
> I am wondering if we should use `ReadOnlyHeaders` instead of
> `ImmutableHeaders` as interface name?
>
> Also, the returned `Header` interface is technically not immutable
> either, because `Header#key()` returns a mutable byte-array... Would we
> need a `ReadOnlyHeader` interface?
>
> If yes, it seems that `ReadOnlyHeaders` should not be a super-interface
> of `Headers` but it would rather be a standalone interface, and a
> wrapper for a `Headers` instance? And `ReadOnlyHeader` would return some
> immutable type instead of `byte[]` for the value()?
>
> An alternative would be to deep-copy the value byte-array what would not
> be free, but given that we are talking about exception handling, it
> would not be on the hot code path, and thus might be acceptable?
>
>
> The above seems to increase the complexity significantly though. Hence,
> I have seconds thoughts on the immutability question:
>
> Do we really need to worry about mutability after all, because in the
> end, KS runtime won't read the Headers instance after the handler was
> called, and if a user modifies the passed in headers, there won't be any
> actual damage (ie, no side effects)? For this case, it might even be ok
> to also not add `ImmutableHeaders` to begin with?
>
>
>
> Sorry for the forth and back (yes, forth and back, because back and
> forth does not make sense -- it's not logical -- just trying to fix
> English :D) as I did bring up the immutability question in the first
> place...
>
>
>
> -Matthias
>
> On 4/25/24 5:56 AM, Loic Greffier wrote:
> > Hi Matthias,
> >
> > I have updated the KIP regarding points 103 and 108.
> >
> > 103.
> > I have suggested a new `ImmutableHeaders` interface to deal with the
> > immutability concern of the headers, which is basically the `Headers`
> > interface without the write accesses.
> >
> > public interface ImmutableHeaders {
> >      Header lastHeader(String key);
> >      Iterable<Header> headers(String key);
> >      Header[] toArray();
> > }
> >
> > The `Headers` interface can be updated accordingly:
> >
> > public interface Headers extends ImmutableHeaders, Iterable<Header> {
> >      //…
> > }
> >
> > Loïc
>