How do you implement request-reply pattern with Spring Kafka?

Spring Kafka's ReplyingKafkaTemplate, a subclass of KafkaTemplate (since 2.1.3), provides request/reply semantics. You configure it with a ProducerFactory and a reply listener container (for example created from a container factory, listening to a reply topic). The template sets three headers by default: KafkaHeaders.CORRELATION_ID to correlate the reply with the request, KafkaHeaders.REPLY_TOPIC to tell the server where to reply, and optionally KafkaHeaders.REPLY_PARTITION. If the reply container listens to a single topic, the template detects the reply topic itself. The server side, for example a @KafkaListener with @SendTo, must echo the correlation id back. Call sendAndReceive(record) (or with a replyTimeout; the default defaultReplyTimeout is 5 seconds), which returns a RequestReplyFuture that is completed asynchronously with the reply or a timeout exception. The caller blocks only if it calls get() on that future, ideally with a timeout. Use waitForAssignment to make sure the reply container is ready before the first request. This gives RPC-style calls over Kafka without an additional HTTP layer.

Expected key concepts: ReplyingKafkaTemplate, reply topic, KafkaHeaders.REPLY_TOPIC, CORRELATION_ID, sendAndReceive, RequestReplyFuture, replyTimeout, reply container, @SendTo, waitForAssignment