kafka springboot

@SpringBootApplication
public class KafkaSpringBootApplication {
    public static void main(String[] args) {
        SpringApplication.run(KafkaSpringBootApplication.class, args);
    }
}

@RestController
public class KafkaController {
    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    @PostMapping("/send")
    public String sendMessage(@RequestParam("message") String message) {
        kafkaTemplate.send("test-topic", message);
        return "Message sent";
    }
}

@KafkaListener(topics = "test-topic", groupId = "group-id")
public void listen(String message) {
    System.out.println("Received message: " + message);
}

spring.kafka.bootstrap-servers=localhost:9092
spring.kafka.consumer.group-id=group-id
spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer
spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer