Package io.smallrye.reactive.messaging.kafka

Interface DeserializationFailureHandler<T>

  • Type Parameters:
    T - the expected type
    public interface DeserializationFailureHandler<T>
    Bean invoked on Kafka deserialization to implement custom failure handling.

    Implementors must use @Identifier to provide a name to the bean. This name is then referenced in the channel configuration: mp.messaging.incoming.my-channel.[key|value]-deserialization-failure-handler=name.

    In case a Kafka Record's key or value cannot be deserialized, this bean can provide failure handling via Mutiny's Uni interface to retry or to provide a fallback value. null is an accepted fallback value. If this bean throws an exception, this is considered as a fatal failure and the application is reported unhealthy.

    • Field Detail

      • DESERIALIZATION_FAILURE_REASON

        static final String DESERIALIZATION_FAILURE_REASON
        Header name for deserialization failure message.
        See Also:
        Constant Field Values

      • DESERIALIZATION_FAILURE_CAUSE

        static final String DESERIALIZATION_FAILURE_CAUSE
        Header name for deserialization failure cause if any.
        See Also:
        Constant Field Values

      • DESERIALIZATION_FAILURE_IS_KEY

        static final String DESERIALIZATION_FAILURE_IS_KEY
        Header name used when the deserialization failure happened on a key. The value is "true" in this case, absent otherwise.
        See Also:
        Constant Field Values

      • DESERIALIZATION_FAILURE_TOPIC

        static final String DESERIALIZATION_FAILURE_TOPIC
        Header name for the topic of the incoming message when a deserialization failure happen.
        See Also:
        Constant Field Values

      • DESERIALIZATION_FAILURE_DATA

        static final String DESERIALIZATION_FAILURE_DATA
        Header name passing the data that was not able to be deserialized.
        See Also:
        Constant Field Values

      • DESERIALIZATION_FAILURE_DESERIALIZER

        static final String DESERIALIZATION_FAILURE_DESERIALIZER
        Header name passing the class name of the underlying deserializer.
        See Also:
        Constant Field Values

      • TRUE_VALUE

        static final byte[] TRUE_VALUE

    • Method Detail

      • handleDeserializationFailure

        default T handleDeserializationFailure​(String topic,
                                       boolean isKey,
                                       String deserializer,
                                       byte[] data,
                                       Exception exception,
                                       org.apache.kafka.common.header.Headers headers)
        Handles a deserialization issue for a record's key or value.
        Parameters:
        topic - the topic
        isKey - whether the failure happened when deserializing a record's key.
        deserializer - the used deserializer
        data - the data that was not deserialized correctly
        exception - the exception
        headers - the record headers, extended with the failure reason, causes, and data. May also be null
        Returns:
        the fallback T

      • decorateDeserialization

        default T decorateDeserialization​(io.smallrye.mutiny.Uni<T> deserialization,
                                  String topic,
                                  boolean isKey,
                                  String deserializer,
                                  byte[] data,
                                  org.apache.kafka.common.header.Headers headers)
        Decorate the given wrapped deserialization action to apply fault tolerance actions. The default implementation calls handleDeserializationFailure(java.lang.String, boolean, java.lang.String, byte[], java.lang.Exception, org.apache.kafka.common.header.Headers) for retro compatibility.
        Parameters:
        deserialization - the deserialization call wrapped in Uni
        topic - the topic
        isKey - whether the deserialization is for a record's key.
        deserializer - the used deserializer
        data - the data to deserialize
        headers - the record headers. May be null
        Returns:
        the recovered T

      • addFailureDetailsToHeaders

        static org.apache.kafka.common.header.Headers addFailureDetailsToHeaders​(String deserializer,
                                                                         String topic,
                                                                         boolean isKey,
                                                                         org.apache.kafka.common.header.Headers headers,
                                                                         byte[] data,
                                                                         Throwable failure)
        Extend record headers to add deserialization failure reason to the incoming message headers
        Parameters:
        deserializer - deserializer class name
        topic - the topic
        isKey - whether the failure happened when deserializing a record's key.
        headers - the record headers
        data - the data that was not deserialized correctly
        failure - the deserilization failure
        Returns:
        headers