Request-Reply and Correlation Patterns

## The problem Kafka is fundamentally a **publish-subscribe log**: a *producer* appends *records* (messages) to a *topic* (a named, partitioned, append-only stream), and *consumers* read records from that topic at their own pace. There is no built-in concept of a 'response' — communication is one-directional and asynchronous (the producer does not wait for anyone to read). **Request-reply** (also called *RPC-style* messaging — Remote Procedure Call, i.e. 'call a function on another service and get a value back') is a pattern that simulates a synchronous call over this async substrate. Service A wants to ask Service B a question and get an answer back. ## What you must add on top of pub/sub 1. **Two topics (or two roles):** a *request topic* that the caller writes to and the responder consumes, and a *reply topic* that the responder writes to and the original caller consumes. The caller tells the responder where to reply by putting the reply-topic name in a header on each request — in Spring Kafka this is the `KafkaHeaders.REPLY_TOPIC` header. 2. **A correlation ID.** Because many requests flow concurrently and Kafka delivers records in bulk, when a reply lands on the reply topic the caller must know *which* outstanding request it answers. The caller stamps each request with a unique **correlation ID** (`KafkaHeaders.CORRELATION_ID`, typically a UUID's bytes). The responder must **copy that exact ID** onto the reply record's header. The caller keeps a **pending-request map** (correlation ID → a future/promise) and, when a reply arrives, looks up the ID and completes the matching future. 3. **A timeout.** Since the responder might be down or slow, the caller cannot wait forever. Each pending request has a timeout; on expiry the future is completed exceptionally (e.g. `KafkaReplyTimeoutException` in Spring) and the entry is evicted from the map. ## In Spring Kafka concretely `ReplyingKafkaTemplate<K,V,R>` wraps a producer plus a `KafkaMessageListenerContainer` on the reply topic. You call `sendAndReceive(record)`; it sets the correlation header, registers the future in an internal `ConcurrentHashMap`, sends the request, and returns a `RequestReplyFuture`. Its listener side receives replies, reads the correlation ID, and completes the corresponding future. The responder side is usually a `@KafkaListener` method that `return`s a value, which Spring routes back to the reply topic carried in the request headers. ## Why not just use this everywhere? Request-reply re-introduces *temporal coupling* (caller blocks waiting) and operational complexity (reply topics, correlation, timeouts) that pub/sub was designed to remove. It is appropriate when a caller genuinely needs the result before proceeding; otherwise plain async events are simpler and more resilient.