Add Kafka Stream exception handlers that push to DLQ

[adrien-ben]

Hi,

It would be nice to have implementations of Kafka Streams' DeserializationExceptionHandler and ProductionExceptionHandler that push to dead-letter queue and enhance the record the way DeadLetterPublishingRecoverer does.

[garyrussell]

This turns out to be a little bit involved because you can't pass live objects in StreamsConfig properties.

I created a DeadLetterPublishingRecoverer as a @Bean so the DeserializationExceptionHandler implementation can invoke it but by the time we get to DeserializationExceptionHandler.configure(), the property has had a toString() operation performed on it 😦 .

It has to be a live object because it needs a template to do the publishing; it would be a pain to have to construct a template from properties in configure().

I will see if I can come up with a reasonable work-around.

[sobychacko]

@garyrussell I actually hit some problems when implementing custom deserialization handlers in the Kafka Streams binder. I ended up doing some manual commits in the DLQ handler that the binder provides. Here are some details about the issues I ran into:
https://stackoverflow.com/questions/48470899/kafka-streams-deserialization-handler
which lead to the creation of this issue in the Kafka jira: https://issues.apache.org/jira/browse/KAFKA-6502

[garyrussell]

@adrien-ben I am not sure what we can do on the producer side. What do you have in mind?

[adrien-ben]

@garyrussell Sorry, I'm no sure I understand what you mean by "what we can do on the producer side".

[garyrussell]

@adrien-ben

It would be nice to have implementations of Kafka Streams' DeserializationExceptionHandler and ProductionExceptionHandler that push to dead-letter queue

If we get a SerializationException on the producer side (handled by a ProductionExceptionHandler), how can we "push to a dead-letter queue"; we would need to serialize the bad data somehow.

Thanks.

[adrien-ben]

Oh yeah sorry, I though you were talking about the producer that pushes to the DLQ.

Actually I'm not even sure that the ProductionExceptionHandler catches serialization exception since the handle method already provides the serialized record. It's not clear in the documentation, I'll dig a bit in the Kafka Stream source and come back to you.

[garyrussell]

I didn't look further, but KIP-210 says it should (at least it was proposed).

[adrien-ben]

From what I see here it seems serialized exception are not handled by the ProductionExceptionHandler.

I'll write a small test application to make sure of it.

[adrien-ben]

There it is

The exception goes all the way back to StreamThread#run()

[garyrussell]

Yes, I see; although productionExceptionIsFatal includes it, even though the serialization is done before the record is created.

    private boolean productionExceptionIsFatal(final Exception exception) {
        final boolean securityException = exception instanceof AuthenticationException ||
            exception instanceof AuthorizationException ||
            exception instanceof SecurityDisabledException;

        final boolean communicationException = exception instanceof InvalidTopicException ||
            exception instanceof UnknownServerException ||
            exception instanceof SerializationException ||
            exception instanceof OffsetMetadataTooLarge ||
            exception instanceof IllegalStateException;

        return securityException || communicationException;
    }

Since all these exceptions are fatal anyway and the productionExceptionHandler is not called for them, I wonder what other types of send exception would make sense to forward the failed send to another topic?

[adrien-ben]

Yes, honnestly I cannot come up with a case where it would make sense to push to DLQ on a production exception. I actually needed a handler that could push to dlq in case of deserialization exception and thought that might be good to ask for the production handler at the same time. But the more I think about it, the less I can make sense of it.