Consumer with Message Headers

@Service
public class HeaderAwareConsumer {

    // Inject specific headers via @Header
    @KafkaListener(topics = "orders.order.created")
    public void onOrder(
        OrderEvent event,
        @Header(KafkaHeaders.RECEIVED_TOPIC) String topic,
        @Header(KafkaHeaders.RECEIVED_PARTITION) int partition,
        @Header(KafkaHeaders.OFFSET) long offset,
        @Header(name = "X-Trace-Id", required = false) String traceId,
        @Header(name = "X-Event-Version", required = false) String eventVersion
    ) {
        MDC.put("traceId", traceId != null ? traceId : "unknown");
        log.info("Processing order from {}-{} @ offset {}", topic, partition, offset);
        // process...
    }
}

// Producer: add headers
@Service
public class TracingOrderProducer {

    public void publish(Order order) {
        var record = new ProducerRecord<>("orders.order.created",
            null, System.currentTimeMillis(), order.getId(),
            new OrderCreatedEvent(order));

        record.headers()
            .add(new RecordHeader("X-Trace-Id", TraceContext.current().toBytes()))
            .add(new RecordHeader("X-Event-Version", "2".getBytes()))
            .add(new RecordHeader("X-Source-Service", "order-service".getBytes()));

        kafkaTemplate.send(record);
    }
}