CODEWITHSUNDEEP
springapache-kafkaapache-kafka-streamsspring-kafka
tushar
tushar

Reputation: 499

Deserialization exception in kafka

I am getting following error while using kafka-streams.

[Kafka Stream] 10:09:26.442 ERROR --- o.a.k.s.e.LogAndFailExceptionHandler: Exception caught     
during Deserialization, taskId: 0_0, topic: t.commodity.promotion, partition: 0, offset: 0
                                    
java.lang.IllegalArgumentException: The class'com.course.kafka.kafkaorder.Broker.Message.PromotionMessage'       
is not in the trusted packages: [java.util, java.lang, com.course.stream.broker.message, 
com.course.stream.broker.message.*]. If you believe this class is safe to deserialize,      
please provide its name. If the serialization is only done by a trusted source, you can also enable        
trust all (*).          

[Kafka Stream] 10:09:26.444 ERROR --- o.a.k.s.p.internals.StreamThread: stream-thread [kafka-stream-7972        
450a-443b-8b7b-007e9fdf8e4c-StreamThread-1] Encountered the following exception during 
processing and the thread is going to shut down: 
                
org.apache.kafka.streams.errors.StreamsException: Deserialization exception handler is set to        
fail upon a deserialization error. If you would rather have the streaming pipeline continue after a         
deserialization error, please set the default.deserialization.exception.handler appropriately.

My code is

@Configuration
public class PromotionJsonSerde {
    
    @Bean
    public KStream<String, PromotionMessage> kStreamPromotionUppercase(StreamsBuilder builder)
    {
        var stringSerde = Serdes.String();
        var jsonSerde = new JsonSerde<>(PromotionMessage.class);

      KStream<String, PromotionMessage> sourceStream = builder.stream("t.commodity.promotion", 
                                Consumed.with(stringSerde, jsonSerde));     
      KStream<String, PromotionMessage> uppercaseStream = sourceStream.mapValues(this::uppercasePromotionCode);

      uppercaseStream.to("t.commodity.promotion-uppercase", Produced.with(stringSerde, jsonSerde));

      return sourceStream;

    }

    private PromotionMessage uppercasePromotionCode(PromotionMessage message)
    {
        return new PromotionMessage(message.getPromotionCode().toUpperCase());
    }
}

Promotion Code

public class PromotionMessage {
    
    private String promotionCode;
     // getters,setters, tostring
}

application.yml

logging:
  pattern:
    console: "[Kafka Stream] %clr(%d{HH:mm:ss.SSS}){faint} %clr(${LOG_LEVEL_PATTERN:%5p}) %clr(---){faint} %clr(%-40.40logger{39}){cyan} %clr(:){faint} %m%n${LOG_EXCEPTION_CONVERSION_WORD:%wEx}"
  
spring:
  main:
    banner-mode: OFF
  kafka:
    listener:
      missing-topics-fatal: false
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer

By seeing the error, I tried to add following code in application.yml file. But still I got
same error.

spring:
     kafka:    
       producer:
         streams:
          properties:
            default.deserialization.exception.handler: org.apache.kafka.streams.errors.LogAndContinueExceptionHandler,org.springframework.kafka.streams.RecoveringDeserializationExceptionHandler
        consumer:
          properties:
                spring:
                    json:
                      trusted:
                        packages: "*"
      cloud:
        stream:
          kafka:
            streams:
              bindings:
               process-in-0.consumer:
                deserializationExceptionHandler: logAndContinue

Why I am getting deserialization error ?
If I check, consumer on terminal , I get the message sent by producer.
Can anyone help me ?

Upvotes: 0

Views: 5259

Answers (1)

Gary Russell
Gary Russell

Reputation: 174799

I see you are creating the serde yourself new JsonSerde<>(PromotionMessage.class); - we automatically add that class' package to the trusted packages; hence

trusted packages: [java.util, java.lang, com.course.stream.broker.message, com.course.stream.broker.message.*]

The property is ignored when you create your own serde. The deserializer is trying to create com.course.kafka.kafkaorder.Broker.Message.PromotionMessage which is in a different package; most likely a different class on the producer.

Add this: ((JsonDeserializer) jsonSerde.deserializer()).setUseTypeHeaders(false); to tell the deserializer to ignore the type information in headers and use the provided fallback type instead.

Upvotes: 2

Related Questions