Я пробовал базовое приложение весенней загрузки kafka, используя этот учебник.
Моя иерархия проекта выглядит так:
SpringBootKafkaApplication.java
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
@SpringBootApplication
public class SpringBootKafkaApplication{
public static void main(String[] args) {
SpringApplication.run(SpringBootKafkaApplication.class, args);
}
}
KafkaController.java
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;
import com.temp.kafka_springboot_pocs.services.Producer;
@RestController
@RequestMapping(value = "/kafka")
public class KafkaController {
private final Producer producer;
@Autowired
public KafkaController(Producer producer) {
this.producer = producer;
}
@PostMapping(value = "/publish")
public void sendMessageToKafkaTopic(@RequestParam("message") String message) {
this.producer.sendMessage(message);
}
}
Consumer.java
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Service;
@Service
public class Consumer {
private final Logger logger = LoggerFactory.getLogger(Consumer.class);
@KafkaListener(topics = "kafka_poc_3", groupId = "group_id")
public void consume(String message) {
logger.info(String.format("$$ -> Consumed Message -> %s", message));
}
}
Producer.java
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;
@Service
public class Producer {
private static final Logger logger = LoggerFactory.getLogger(Producer.class);
private static final String TOPIC = "kafka_poc_3";
@Autowired
private KafkaTemplate<String, String> kafkaTemplate;
public void sendMessage(String message) {
logger.info(String.format("$$ -> Producing message --> %s",message));
this.kafkaTemplate.send(TOPIC, message);
}
}
application.properties
spring.kafka.bootstrap-servers=localhost:9092
spring.kafka.consumer.group-id=group-1
spring.kafka.consumer.auto-offset-reset=earliest
spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserialize
spring.kafka.producer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
spring.kafka.producer.value-deserializer=org.apache.kafka.common.serialization.StringDeserialize
Сбой при загрузке приложения Springначать со следующего описания ошибки:
Описание:
Failed to bind properties under 'spring.kafka.consumer.value-deserializer' to java.lang.Class<?>:
Property: spring.kafka.consumer.value-deserializer
Value: org.apache.kafka.common.serialization.StringDeserialize
Origin: class path resource [application.properties]:12:42
Reason: No converter found capable of converting from type [java.lang.String] to type [java.lang.Class<?>]
Что мне здесь не хватает?
PS: Весной и весной я долго не работал. Так будет любить кого-то, дающего мне хорошее руководство / связь с примером или точным ответом.