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
