Support `@SendTo` (replying) for log-compacted topics

[nyilmaz]

Currently MessagingMessageListenerAdapter#sendResponse(Object result, String topic, @Nullable Object source) method is not setting key before sending and causes CorruptRecordException as stated here.

Workaround is to write a ProducerInterceptor and add it to producer config like;

extended.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, List.of(KeySetterProducerInterceptor.class));

Planning to support this?

[garyrussell]

The problem is there is only a single result; we'd need some expression to evaluate to determine what to set the key to.

We'd need another @SendTo annotation since the current one is owned by spring-messaging (or we'd need a change there).

There currently is another workaround - return a message...

@SpringBootApplication
public class Kgh725Application {

	public static void main(String[] args) {
		SpringApplication.run(Kgh725Application.class, args);
	}

	@Bean
	public ApplicationRunner runner(KafkaTemplate<String, String> template) {
		return args -> {
			template.send("kgh725-1", "foo", "bar");
		};
	}

	@KafkaListener(id = "one", topics = "kgh725-1")
	@SendTo("kgh725-2")
	public Message<?> listen(String in) {
		return MessageBuilder.withPayload(in.toUpperCase())
				.setHeader(KafkaHeaders.TOPIC, "kgh725-2")
				.setHeader(KafkaHeaders.MESSAGE_KEY, "baz")
				.build();
	}

	@KafkaListener(id = "two", topics = "kgh725-2")
	public void listen(ConsumerRecord<String, String> in) {
		System.out.println(in);
	}

	@Bean
	public NewTopic topic1() {
		return new NewTopic("kgh725-1", 1, (short) 1);
	}

	@Bean
	public NewTopic topic2() {
		return new NewTopic("kgh725-2", 1, (short) 1);
	}

}

(Note that you have to populate the topic header too for this work (it's also needed in the @SendTo).

Result:

ConsumerRecord(topic = kgh725-2, partition = 0, offset = 0, CreateTime = 1530812067965, serialized key size = 3, serialized value size = 3, headers = RecordHeaders(headers = [], isReadOnly = false), 
key = baz, value = BAR)

[nyilmaz]

Is this approach valid for batch message listeners?

[garyrussell]

Not really; only one reply message will be sent for the whole batch; we should probably look for a Collection<Message<?>> in the return type to support that scenario.

[nyilmaz]

Well, i don't know what would be your precedence of this issue (and #726), but writing a type-safe ProducerInterceptor configurer (maybe a simple setter) for ProducerFactory and documenting it, covers single and batch record listeners altogether and would be enough for most usecases imho.

[garyrussell]

It's not clear to me how that can intelligently set the key since the interceptor only has access to the ProducerRecord. It would only work if the key can be derived from the value. For example, people might want to compute the key based on the key in the request record.