Use the original key in reply messages when using ReplyingKafkaTemplate and SendTo

[neshkeev]

Expected Behavior

Response messages in addition to the kafka_correlationId Header should have the same keys as their corresponding request messages when constructing the Request-Reply semantics on top of ReplyingKafkaTemplate and SendTo.

Current Behavior

The response messages have no keys (null)

Context

When one sends a message/request it's only natural to expect that the response message have same key as the original message (request). Such an explicit formalization can allow engineers to reason about the workload. Also it might come in handy when there are different consumers listening the reply topic.

Here is a brief example of what this issue is about

Sender:

void sender(ReplyingKafkaTemplate<String, String, String> template) {
    final ProducerRecord<String, String> record = new ProducerRecord<>(
            "mytopic",
            "key",
            "value"
    );

    var replay = template.sendAndReceive(record, responseTimeout);

    // THIS ASSERTION FAILS
    assert "key".equals(replay.get().key()) : "The response should have the same key";
}

Listener:

@SendTo
@Listener(topics="mytopic")
String listen(String value) {
    return "Hello, " + value;
}

If you run this snippet, you will get the assertion error, because the key of the replying message is null.

[artembilan]

This makes sense.
Feel free to contribute the fix: https://github.com/spring-projects/spring-kafka/blob/main/CONTRIBUTING.adoc
Thanks

[sobychacko]

I am taking a look at this issue. It's straightforward with a single return record, as in the example above. I wonder what the semantics are if the return type is a batch or a collection. In the case of a collection, the framework may not detect what keys to provide, and it is up to the application to provide the keys by wrapping them in a Message. What do you think?