Kafka使用者使用ClassNotFoundException

Question

我有一个用Spring Boot 2.0.2编写的Kafka客户应用程序。当我在侦听器中收到消息时，出现以下错误：

Caused by: org.springframework.messaging.converter.MessageConversionException: failed to resolve class name. Class not found [com.test.demo.domain.Account]; nested exception is java.lang.ClassNotFoundException: com.test.demo.domain.Account

生产者中对象的类名称是“ com.test.demo.domain.Account ”，但是在使用者中我有不同的包和类名称。

当我重新包装消费者的类名以匹配生产者时，一切正常。但是，我相信我不必这样做。

有人知道这个问题吗？

====更新====

我的生产者代码：

@Bean public ProducerFactory<String, Account> accountProducerFactory() {
    Map<String, Object> configProps = new HashMap<>();
    configProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapAddress);
    configProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
    configProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class.getName());
    return new DefaultKafkaProducerFactory<>(configProps); }

@Bean public KafkaTemplate<String, Account> accountKafkaTemplate() {
    ProducerFactory<String, Account> factory = accountProducerFactory();

    return new KafkaTemplate<>(factory); }

消费者代码：

public ConsumerFactory<String, Account> accountConsumerFactory() {
    Map<String, Object> configProps = new HashMap<>();
    configProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapAddress);
    configProps.put(ConsumerConfig.GROUP_ID_CONFIG, groupName);
    configProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
    configProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, BytesDeserializer.class.getName());
    return new DefaultKafkaConsumerFactory<>(configProps);
}

@Bean
public KafkaListenerContainerFactory<?> kafkaJsonListenerContainerFactory() {
    ConcurrentKafkaListenerContainerFactory<String, Account> factory = new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(accountConsumerFactory());
    factory.setMessageConverter(new StringJsonMessageConverter());
    return factory;
}

例外：

org.springframework.kafka.listener.ListenerExecutionFailedException: Listener method could not be invoked with the incoming message
Endpoint handler details:
Method [public void com.kconsumer.accountconsumer.service.AccountConsumer.accountListener(com.kconsumer.accountconsumer.domain.Account)]
Bean [com.kconsumer.accountconsumer.service.AccountConsumer@444cc791]; nested exception is org.springframework.messaging.converter.MessageConversionException: Cannot handle message; nested exception is org.springframework.messaging.converter.MessageConversionException: Cannot convert from [java.lang.String] to [com.kconsumer.accountconsumer.domain.Account] for GenericMessage [payload={"id":"5b079d0b340d9ef2ac9b6f02","name":"test-400","version":0}, headers={kafka_offset=0, kafka_consumer=org.apache.kafka.clients.consumer.KafkaConsumer@6a0a6b0b, kafka_timestampType=CREATE_TIME, kafka_receivedMessageKey=5b079d0b340d9ef2ac9b6f02, kafka_receivedPartitionId=0, kafka_receivedTopic=ktest, kafka_receivedTimestamp=1527225611820, __TypeId__=[B@2f92e17a}], failedMessage=GenericMessage [payload={"id":"5b079d0b340d9ef2ac9b6f02","name":"test-400","version":0}, headers={kafka_offset=0, kafka_consumer=org.apache.kafka.clients.consumer.KafkaConsumer@6a0a6b0b, kafka_timestampType=CREATE_TIME, kafka_receivedMessageKey=5b079d0b340d9ef2ac9b6f02, kafka_receivedPartitionId=0, kafka_receivedTopic=ktest, kafka_receivedTimestamp=1527225611820, __TypeId__=[B@2f92e17a}]
    at org.springframework.kafka.listener.adapter.MessagingMessageListenerAdapter.invokeHandler(MessagingMessageListenerAdapter.java:257)

Answer 1

如果使用@KafkaListener，请使用StringDeserializer或a ByteArrayDserializer并将a添加StringJsonMessageConverter @Bean到应用程序上下文中。

然后...

@KafkaListener(...)
public void listen(Account account) {
    ...
}

...所需的帐户类型将传递给转换器。

请参阅文档。

编辑

您不需要连接工厂，引导程序将检测该转换器并为您连接。

这是一个例子：

@SpringBootApplication
public class So50478267Application {

    public static void main(String[] args) {
        SpringApplication.run(So50478267Application.class, args);
    }

    @Bean
    public ApplicationRunner runner(KafkaTemplate<String, Account1> template) {
        return args -> template.send("so50478267", new Account1("foo.inc"));
    }

    @KafkaListener(id = "listen", topics = "so50478267")
    public void listen(Account2 account) {
        System.out.println(account);
    }

    @Bean
    public StringJsonMessageConverter jsonConverter() {
        return new StringJsonMessageConverter();
    }

    @Bean
    public NewTopic topic() {
        return new NewTopic("so50478267", 1, (short) 1);
    }

    public static class Account1 {

        private final String customer;

        public Account1(String customer) {
            this.customer = customer;
        }

        public String getCustomer() {
            return this.customer;
        }

    }

    public static class Account2 {

        private String customer;

        public Account2() {
            super();
        }

        public String getCustomer() {
            return this.customer;
        }

        public void setCustomer(String customer) {
            this.customer = customer;
        }

        @Override
        public String toString() {
            return "Account2 [customer=" + this.customer + "]";
        }

    }

}

和

spring.kafka.consumer.auto-offset-reset=earliest
spring.kafka.consumer.enable-auto-commit=false

spring.kafka.producer.value-serializer=org.springframework.kafka.support.serializer.JsonSerializer