How do you use Kafka headers in Spring Kafka?

Producer: kafkaTemplate.send(MessageBuilder.withPayload(value).setHeader('traceId', traceId).setHeader(KafkaHeaders.TOPIC, 'my-topic').build()). When you send a Message<?>, topic, partition, key and timestamp are carried in headers (KafkaHeaders.TOPIC, PARTITION, KEY, TIMESTAMP) and the payload is the data. Consumer: a @KafkaListener method can take @Header('traceId') String traceId as a parameter, standard ones such as @Header(KafkaHeaders.RECEIVED_TOPIC) and RECEIVED_PARTITION, or the ConsumerRecord<K,V>, whose record.headers() exposes the Kafka headers. In the Kafka client a Header has a String key and a byte[] value (the value may be null), so encode and decode values yourself. Common uses are distributed tracing (pass trace IDs), schema version or routing information. The DeadLetterPublishingRecoverer copies the failed record's original headers to the record it publishes and adds DLT_ headers such as DLT_ORIGINAL_TOPIC, DLT_ORIGINAL_PARTITION and DLT_ORIGINAL_OFFSET plus exception details, and retry-topic publishing also adds retry information headers.

Expected key concepts: MessageBuilder, setHeader, KafkaHeaders.TOPIC, @Header, ConsumerRecord, record.headers, byte[], traceId, DeadLetterPublishingRecoverer, DLT_ORIGINAL_TOPIC, distributed tracing