apache-kafka - 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

Producer 中对象的类名是“ 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;
}

异常(exception):

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或 ByteArrayDserializer并添加一个 StringJsonMessageConverter @Bean到应用程序上下文。

然后...

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

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

见 the documentation .

编辑

您不需要连接工厂，启动将检测转换器并为您连接。

下面是一个例子:

@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