Basic Producer with KafkaTemplate

@Service
@Slf4j
public class OrderProducer {

    private final KafkaTemplate<String, OrderEvent> kafkaTemplate;

    public OrderProducer(KafkaTemplate<String, OrderEvent> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    public void publishOrderCreated(Order order) {
        OrderEvent event = OrderEvent.builder()
            .orderId(order.getId())
            .status("CREATED")
            .timestamp(Instant.now())
            .build();

        kafkaTemplate.send("orders.order.created", order.getId(), event)
            .whenComplete((result, ex) -> {
                if (ex != null) {
                    log.error("Failed to publish order event: orderId={}", order.getId(), ex);
                } else {
                    log.info("Published order event: orderId={}, offset={}",
                        order.getId(), result.getRecordMetadata().offset());
                }
            });
    }

    // Synchronous send (blocks until ack received)
    public void publishSync(String topic, String key, Object value) {
        try {
            SendResult<String, Object> result = kafkaTemplate.send(topic, key, value).get();
            log.info("Message sent successfully to partition {} offset {}",
                result.getRecordMetadata().partition(),
                result.getRecordMetadata().offset());
        } catch (Exception e) {
            throw new RuntimeException("Failed to send message to Kafka", e);
        }
    }
}