CODEWITHSUNDEEP
javaspringspring-bootapache-kafkajson-deserialization
Punter Vicky
Punter Vicky

Reputation: 16992

Spring Kafka Custom Deserializer

I am following the steps listed in this link to create a customer deserializer. The message that I receive from Kafka has plain text "log message -" before the json string.I want the deserializer to ignore this string and parse the json data. Is there a way to do it?

Application

@SpringBootApplication
public class TransactionauditServiceApplication {

    public static void main(String[] args) throws InterruptedException {
        new SpringApplicationBuilder(TransactionauditServiceApplication.class).web(false).run(args);
    }

    @Bean
    public MessageListener messageListener() {
        return new MessageListener();
    }

    public static class MessageListener {

        @KafkaListener(topics = "ctp_verbose", containerFactory = "kafkaListenerContainerFactory")
        public void listen(@Payload ConciseMessage message, 
                  @Header(KafkaHeaders.RECEIVED_PARTITION_ID) int partition) {
            System.out.println("Received Messasge in group foo: " + message.getStringValue("traceId") + " partion " + partition);
        }
    }
}

ConsumerConfig

@EnableKafka
@Configuration
public class KafkaConsumerConfig {

    @Value(value = "${kafka.bootstrapAddress:localhost:9092}")
    private String bootstrapAddress;

    @Value(value = "${groupId:audit}")
    private String groupId;

    @Bean
    public ConsumerFactory<String, ConciseMessage> consumerFactory() {
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapAddress);
        props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
        return new DefaultKafkaConsumerFactory<>(props, new StringDeserializer(), new JsonDeserializer<>(ConciseMessage.class));
    }

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, ConciseMessage> kafkaListenerContainerFactory() {

        ConcurrentKafkaListenerContainerFactory<String, ConciseMessage> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        return factory;
    }
}

Upvotes: 2

Views: 8914

Answers (1)

adarshr
adarshr

Reputation: 62583

By writing the line new JsonDeserializer<>(ConciseMessage.class), you are just telling Kafka that you want to convert the message to a ConciseMessage type. So, that doesn't make it a custom deserializer. To fix your problem you will most likely have to come up with your own implementation of a deserializer that has the logic to strip the text "log message -".

Upvotes: 2

Related Questions