由于“无法从 [[B] 转换为 [com.myClass] for GenericMessage [payload info]”导致的传入消息无法调用侦听器方法

Question

我正在尝试学习 Kafka + Springboot。我想在我的消费者类中添加第二个消费者，它订阅与第一个主题相同的主题，但具有不同的 groupID。当我只有第一个消费 Json 的消费者（或者至少输出是 Json？）时，这些类不是很复杂并且可以工作。还要解释一下，我只从一个使用 @EnableBindings 方法的生产者和消费者开始，但它已被弃用，因此我正在学习正确/新的做法。

任何提示appreicated！请让我走上正确的道路。

我有很多 Maven 依赖，所以我简单总结一下：它包括 spring-kafka、kafka-streams、spring-boot-starter-jpa 等等......

应用程序属性，我不确定底部的标题属性是否正确：

spring.kafka.bootstrap-servers=localhost:29092
spring.kafka.consumer.properties.spring.json.trusted.packages=*
spring.kafka.consumer.auto-offset-reset=earliest

spring.kafka.consumer.properties.key-deserializer=org.apache.kafka.common.serialization.ErrorHandlingDeserializer

spring.kafka.consumer.properties.value-deserializer=org.springframework.kafka.support.serializer.ErrorHandlingDeserializer

spring.kafka.consumer.properties.spring.deserializer.key.delegate.class: org.apache.kafka.common.serialization.StringDeserializer
spring.kafka.consumer.properties.spring.deserializer.value.delegate.class: org.springframework.kafka.support.serializer.JsonDeserializer

spring.kafka.producer.properties.key-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.properties.value-serializer=org.springframework.kafka.support.serializer.JsonSerializer

spring.kafka.producer.properties.spring.json.add.type.headers=false
spring.kafka.consumer.properties.spring.json.use.type.headers=false

#cockroachDB configs omitted

消费类：

@Service
public class BookConsumer {
    
    @Autowired
    public BookConsumer(BookService bookService) {
        this.bookService=bookService;
    }
    
    private final BookService bookService;
    
    @KafkaListener(topics="testKafka", groupId="group2")
    public void consume(BookDto books) {
        System.out.println("saved!");
        bookService.save(books);
    }
    
    @KafkaListener(topics="testKafka", groupId="group_id")
    public void consumeMessage(BookDto message){
        System.out.println(message);
    }
}

生产者类：

@Service
public class BookProducer {
    
    @Autowired
    private KafkaTemplate<String,BookDto> jsonTemplate;
    
    
    public void sendBookEvent(BookDto book) {
        this.jsonTemplate.send("testKafka", book);
    }
    
    public void sendJson(BookDto booklist) {
        this.jsonTemplate.send("testKafka", booklist);
    }
    
}

我还有一个调用事物的 Restcontroller，我只包括与生产者和消费者相关的两个。它是“/sendBookFromList”，应该用于当前不工作的消费者：

@RestController
public class HelloController {
    
    private final BookProducer producer;
    
    @Autowired
    private final BookService bookService;
    
    @Autowired
    public HelloController(BookProducer producer, BookService bookService) {
        this.producer=producer;
        this.bookService=bookService;
    }
    

    public List<BookDto> makeList() {
        List<BookDto> readingList = new ArrayList<BookDto>();
        readingList.add(new BookDto(1, "Needful Things", "Stephen King"));
        readingList.add(new BookDto(2, "The Three-Body Problem", "Liu Cixin"));
        readingList.add(new BookDto(666, "Cujo", "Stephen King"));
        readingList.add(new BookDto(8, "The Castle", "Franz Kafka"));
        return readingList;
    }

    @RequestMapping("json/{pos}")
    public String sendJson(@PathVariable("pos") Integer pos) {
        producer.sendJson(makeList().get(pos));
        return "BookDto sent!";
    }

    @RequestMapping("/sendBookFromList/{listPos}")
    public String sendBooks(@PathVariable("listPos") Integer pos) {
        producer.sendBookEvent(makeList().get(pos));
        return "added!";
    }

我有一个 BookDto 类和一个实体，因为我将它连接到一个 cockroachDB，我将包括在内，以防万一：

public class BookDto {

    private Integer id;
    private String name;
    private String Author;
    
    public BookDto() {
        
    }

    public BookDto(Integer id, String name, String Author) {
        this.id = id;
        this.name = name;
        this.Author = Author;
    }

//I'll omit the getter and setters here but they exist!
    
      @Override public String toString() {
          return "Book "+id+": "+name+" by "+Author; }   
}

//I'm using Lombok as well, I didn't forget my constructors and stuff I swear!
@Entity(name="BOOK")
@Data
public class Book {
    
    @Id 
    private Integer id;
    
    private String name;
    private String author;

}

为了澄清，我正在使用 Mapper，因为我认为这可能是从 Dto 和实体转换之间的问题。我认为它不起作用，因为这是错误消息（以前是 Book 而不是没有映射器的 BookDto）：

Listener method could not be invoked with the incoming message
Endpoint handler details:
Method [public void com.Book.kafka.BookConsumer.consume(com.Book.kafka.BookDto)]

Cannot convert from [[B] to [com.Book.kafka.BookDto] for GenericMessage [payload=byte[48], headers={kafka_offset=151, kafka_consumer=org.apache.kafka.clients.consumer.KafkaConsumer@1ce9bcc9, kafka_timestampType=CREATE_TIME, kafka_receivedPartitionId=0, kafka_receivedTopic=testKafka, kafka_receivedTimestamp=1649930203804, __TypeId__=[B@163eece4, kafka_groupId=group2}]

附加信息：我在 docker 中运行 Kafka 和 Zookeeper

Answer 1

非常愚蠢的解决方案，但它似乎解决了我的问题。我怀疑我搞砸了属性，我是对的！ （我认为）

从应用程序属性中的所有spring.kafka.[consumer/producer].properties.[value/key]-deserializer 中删除.properties 后，我启动并运行良好。

我不知道为什么要修复它，但消费者现在可以正确处理它并将我的书项保存到 cockroachDB。

Answer 2

我遇到了同样的问题，并在这个平台上的另一个问题上找到了这个解决方案： Kafka Listener method could not be invoked with the incoming message

希望对你有所帮助。