Исключение при обработке схемы реестра десериализатором кафки - PullRequest
Новая
       35

Исключение при обработке схемы реестра десериализатором кафки

0 голосов
/ 16 марта 2019

Я перехожу по сливной ссылке , чтобы публиковать сообщения kafka об изменениях в таблице mysql. Когда я пытаюсь использовать это сообщение из приложения Springboot, я получаю следующее исключение. Как я могу это исправить, чтобы я мог прочитать сообщение. Иногда я могу прочитать сообщение, но получаю сериализованную версию, например consumed message is key: null value : 2foo��䶰Z��䶰Z

Caused by: java.lang.ClassCastException: java.lang.String cannot be cast to com.spring.kafkaexample.springbootkafkaconsumer.model.Foobar
    at com.spring.kafkaexample.springbootkafkaconsumer.listener.KafkaConsumer.consume(KafkaConsumer.java:22) ~[classes/:na]
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) ~[na:1.8.0_131]
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) ~[na:1.8.0_131]
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) ~[na:1.8.0_131]
    at java.lang.reflect.Method.invoke(Method.java:498) ~[na:1.8.0_131]
    at org.springframework.messaging.handler.invocation.InvocableHandlerMethod.doInvoke(InvocableHandlerMethod.java:181) ~[spring-messaging-5.0.8.RELEASE.jar:5.0.8.RELEASE]
    at org.springframework.messaging.handler.invocation.InvocableHandlerMethod.invoke(InvocableHandlerMethod.java:114) ~[spring-messaging-5.0.8.RELEASE.jar:5.0.8.RELEASE]

Ниже мой код 

@KafkaListener(topics = "mysql-foobar", groupId = "group_id")
public void consume(ConsumerRecord<String, Foobar> message) {

    System.out.println("consumed message is key: "+message.key() + " value :  "+message.value());
}

десериализатор 

@Bean
public Map<String, Object> consumerConfigs() {
    Map<String, Object> props = new HashMap<>();
    props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");
    props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, KafkaAvroDeserializer.class);
    props.put(ConsumerConfig.GROUP_ID_CONFIG, "avro");

    return props;
}

1 Ответ

0 голосов
/ 16 марта 2019

Если вы получаете String cannot be cast to com.spring.kafkaexample.springbootkafkaconsumer.model.Foobar, то это означает, что у вас есть продюсер, отправляющий строку в вашу тему, что недопустимо для создания экземпляра Foobar.

Если вы получаете материал 2foo��䶰Z��䶰Z, это будет означать, что вы распечатываете необработанные байты UTF-8 некоторых двоичных данных, таких как Avro, и вам лучше было бы извлечь поляобъект, а не полагаться исключительно на неявный метод toString.

...