I'm using kafka for my api. I'm using spring with microservice I'll post my kafka code below:

Command:

private static final Logger logger = 
        LoggerFactory.getLogger(UserCommandServiceImpl.class);

@Autowired
private KafkaTemplate<String, Object> kafkaTemplate;

public void sendMessage(User objeto) 
{
    
    logger.info(String.format("Message sent -> %s", objeto.toString()));
    this.kafkaTemplate.send("quickstart-events", objeto);

}

Query:

private final Logger logger = LoggerFactory.getLogger(UserQueryServiceImpl.class);

   
@Autowired
private MongoTemplate mongoTemplate;

@KafkaListener(topics = "quickstart-events" , groupId = "group-id")
public void consume(String message) 
{
    logger.info(String.format("Message recieved -> %s", message));
    mongoTemplate.insert(message, "user");
    
}

I installed kafka from that