Spring Kafka integration: cách produce và consume message?

// Producer
@Service
class OrderEventPublisher {
  private final KafkaTemplate<String, OrderEvent> kafkaTemplate;

  public void publish(Order order) {
    kafkaTemplate.send("order-events", order.getId().toString(),
        new OrderEvent(order.getId(), "ORDER_CREATED"))
      .whenComplete((result, ex) -> { if (ex != null) log.error("Send failed", ex); });
  }
}

// Consumer
@KafkaListener(topics = "order-events", groupId = "inventory-service")
public void handle(OrderEvent event) {
  inventoryService.reserve(event.getOrderId());
}