KafkaTemplate and ProducerFactory Configuration

@Service
@RequiredArgsConstructor
public class OrderEventPublisher {

    private final KafkaTemplate<String, OrderEvent> template;

    // Async with callback (recommended)
    public void publish(OrderEvent event) {
        template.send("orders", event.orderId().toString(), event)
            .whenComplete((result, ex) -> {
                if (ex != null) log.error("Send failed for {}", event.orderId(), ex);
                else log.info("Sent to {} offset {}",
                    result.getRecordMetadata().topic(),
                    result.getRecordMetadata().offset());
            });
    }

    // Sync (only when you must block — e.g. on app shutdown)
    public void publishSync(OrderEvent event) throws ExecutionException, InterruptedException {
        template.send("orders", event.orderId().toString(), event)
            .get(5, TimeUnit.SECONDS);
    }
}

// Custom factory when you need a second producer (e.g. transactional)
@Configuration
class TxProducerConfig {
    @Bean
    public ProducerFactory<String, byte[]> txProducerFactory(KafkaProperties props) {
        Map<String, Object> cfg = new HashMap<>(props.buildProducerProperties());
        cfg.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "tx-orders-" + UUID.randomUUID());
        cfg.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
        DefaultKafkaProducerFactory<String, byte[]> f = new DefaultKafkaProducerFactory<>(cfg);
        f.setTransactionIdPrefix("tx-orders-");
        return f;
    }
    @Bean public KafkaTemplate<String, byte[]> txTemplate(ProducerFactory<String, byte[]> f) {
        return new KafkaTemplate<>(f);
    }
}