Listener method could not be invoked with the incoming message caused by "Cannot convert from [[B] to [com.myClass] for GenericMessage [payload info]"

Question

I'm trying to learn Kafka + Springboot. I wanted to add a second consumer in my consumer class that subscribes to the same topic as the first one but has a different groupID. The classes are not very complicated and works when I only have the first consumer that consumes Json (or at least the output is Json?). Also to explain a bit, I started with only one producer and consumer that was using the @EnableBindings method but it's deprecated so I'm learning the right/new way of doing it.

Any tips appreicated! Please put me on the right path.

I have a lot of Maven dependencies so I will just summarize: it includes spring-kafka, kafka-streams, spring-boot-starter-jpa among others....

application properties, I'm not sure if the headers properties at the bottom are even correct:

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

Consumer class:

@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);
    }
}

Producer class:

@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);
    }
    
}

I also have a Restcontroller that invokes things, I'll only include the two that are related to the producer and consumer. It's the "/sendBookFromList" one that should be used for the consumer that is currently not working anyway:

@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!";
    }

I have a BookDto class as well as an Entity because I have it connected to a cockroachDB that I will include just in case:

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;

}

for clarification I'm using a Mapper because I thought that might be the problem between a conversion from Dto and Entity. I think it doesn't work because this is the Error message (used to be Book instead of BookDto without the mapper):

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}]

Additional info: I'm running Kafka and Zookeeper in docker

Answer 1

Very dumb solution but it seemed to solve my problem. I suspected that I messed up the properties and I was right! (I think)

I got it up and running just fine after removing the .properties from all spring.kafka.[consumer/producer].properties.[value/key]-deserializer within the application properties.

I don't know WHY this fixed it, but the consumer can now process it properly and save my book item to cockroachDB.