Отправка с использованием KafkaTemplate <String, String> возвращает исключение для использования String в качестве ключа - PullRequest
Новая
       20

Отправка с использованием KafkaTemplate <String, String> возвращает исключение для использования String в качестве ключа

0 голосов
/ 03 января 2019

При использовании Integer в качестве ключа это не проблема, kafka должен иметь возможность обрабатывать строки как ключ.

ProducerFactory<String, String> pf =
                new DefaultKafkaProducerFactory<String, String>(senderProps);
KafkaTemplate<String, String> template = new KafkaTemplate<>(pf);
ProducerRecord<String,String> pr = new ProducerRecord<>("my-topic", "key1","test");
        template.send(pr);`

Выдает следующее исключение:

Org.apache.kafka.common.errors.SerializationException: Невозможно преобразовать ключ класса java.lang.String в класс org.apache.kafka.common.serialization.IntegerSerializer, указанный в key.serializer

Причина: java.lang.ClassCastException: java.lang.String не может быть приведена к java.lang.Integer в org.apache.kafka.common.serialization.IntegerSerializer.serialize (IntegerSerializer.java:21) в org.apache.kafka.common.serialization.ExtendedSerializer $ Wrapper.serialize (ExtendedSerializer.java:65) в org.apache.kafka.common.serialization.ExtendedSerializer.$ Wrapper.serialize (ExtendedSerializer.java:55) в org.apache.kafka.clients.producer.KafkaProducer.doSend (KafkaProducer.java:799) в org.apache.kafka.clients.producer.KafkaProducer.send (KafkaProducer.java: 784) в org.springframework.kafka.core.DefaultKafkaProducerFactory $ CloseSafeProducer.send (DefaultKafkaProducerFactory.java:285) в org.springframework.kafka.core.KafkaTemplate.doSend (KafkaTemplate.java:357) в org.springframework.kafka.core.KafkaTemplate.send (KafkaTemplate.java:206)

1 Ответ

0 голосов
/ 03 января 2019

Кажется, ваш сериализатор ключей установлен как Integer в ProducerConfig.Вам нужно установить KEY_SERIALIZER_CLASS_CONFIG как String: 

@Bean
    public Map<String, Object> producerConfigs() {
        Map<String, Object> props = new HashMap<>();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        return props;
    }
...