Handle multiple responses with ReplyingKafkaTemplate

[gkohen]

🎁 Enhancement
Continued thread from Stackoverflow.
I'm trying to implement a reply response pattern in which I publish a message to a topic listened to by several consumer groups. That means that they will all get the message as well as submit the response in the reply topic.

The problem is because they all respond to the same message, only the first received message in the reply topic shall be answered. The others will be discarded. Given that I know how many responses I should be getting on the reply topic(call that number-n), how can I make ReplyingKafkaTemplate wait for n responses and then resolve the answer? I've tried inferring from the Spring Kafka documentation but could not quite figure it out

There were few options suggested amongst one is to extend ReplyingKafkaTemplate to support the feature or implement a timer after which you can return partial responses.
At this point ReplyingKafkaTemplate is not really built for extensions and might require almost wholesale copy of the whole class rather than extending it.

[garyrussell]

I thought about it some more, and wonder why something like this wouldn't work...

public class AggregatingReplyingKafkaTemplate<K, V, R> extends ReplyingKafkaTemplate<K, V, Collection<R>> {

	public AggregatingReplyingKafkaTemplate(ProducerFactory<K, V> producerFactory,
			GenericMessageListenerContainer<K, Collection<R>> replyContainer) {

		super(producerFactory, replyContainer);
	}

	@Override
	public void onMessage(List<ConsumerRecord<K, Collection<R>>> data) {
		List<ConsumerRecord<K, Collection<R>>> completed = new ArrayList<>();
		data.forEach(record -> {
			// We should move the following to a helper method
			Iterator<Header> iterator = record.headers().iterator();
			CorrelationKey correlationId = null;
			while (correlationId == null && iterator.hasNext()) {
				Header next = iterator.next();
				if (next.key().equals(KafkaHeaders.CORRELATION_ID)) {
					correlationId = new CorrelationKey(next.value());
				}
			}
			if (correlationId == null) {
				this.logger.error("No correlationId found in reply: " + record
						+ " - to use request/reply semantics, the responding server must return the correlation id "
						+ " in the '" + KafkaHeaders.CORRELATION_ID + "' header");
			}
			// end of helper method

			else {
				List<R> list = addToCollection(record, correlationId);
				if (list.size() == 5) {
					ConsumerRecord<K, Collection<R>> done = new ConsumerRecord<>(record.topic(), record.partition(),
							record.offset(), record.key(), list);
					// add correlation header.
				}
			}
		});
		if (completed.size() > 0) {
			super.onMessage(completed);
		}
	}

	@SuppressWarnings({ "rawtypes", "unchecked" })
	private List<R> addToCollection(ConsumerRecord record, CorrelationKey correlationId) {
		R value = (R) record.value();
		// TODO
		return null;
	}


}

Need to give some thought to how/when to commit offsets, though - could be tricky.

[garyrussell]

I am proposing that we close this as "Won't fix" because the functionality is already available using Spring Integration's Aggregator...

@SpringBootApplication
public class Kgh998Application {

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

	@Bean
	public IntegrationFlow reply1(ConsumerFactory<String, String> cf) {
		return IntegrationFlows.from(Kafka.messageDrivenChannelAdapter(cf, new ContainerProperties("kgh998reply-1"))
					.configureListenerContainer(cont -> cont.groupId("one")))
				.log()
				.channel("aggregate.input")
				.get();
	}

	@Bean
	public IntegrationFlow reply2(ConsumerFactory<String, String> cf) {
		return IntegrationFlows.from(Kafka.messageDrivenChannelAdapter(cf, new ContainerProperties("kgh998reply-2"))
				.configureListenerContainer(cont -> cont.groupId("two")))
				.log()
				.channel("aggregate.input")
				.get();
	}

	@Bean
	public IntegrationFlow aggregate() {
		return f -> f.aggregate(a -> a
					.correlationExpression("new String(headers.kafka_correlationId)")
					.releaseExpression("size() == 2"))
				.handle(System.out::println);
	}

	@KafkaListener(id = "server1", topics = "kgh998")
	@SendTo("kgh998reply-1")
	public String listen1(String in) {
		System.out.println("1:" + in);
		return in.toUpperCase();
	}

	@KafkaListener(id = "server2", topics = "kgh998")
	@SendTo("kgh998reply-2")
	public String listen2(String in) {
		System.out.println("2:" + in);
		return in.toLowerCase();
	}

	@Bean
	public ApplicationRunner runner(KafkaTemplate<String, String> template) {
		return args -> {
			ProducerRecord<String, String> record = new ProducerRecord<>("kgh998", "TesT");
			record.headers()
					.add(new RecordHeader(KafkaHeaders.CORRELATION_ID, UUID.randomUUID().toString().getBytes()));
			template.send(record);
		};
	}

	@Bean
	public NewTopic topic() {
		return new NewTopic("kgh998", 1, (short) 1);
	}

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

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

}

and

1:TesT
2:TesT
GenericMessage [payload=[test, TEST], headers=...

[gkohen]

@garyrussell, I understand some of it might be already available somewhere else in the Spring ecosystem. Is the requirement to use yet another spring library just for this, a burden for users having to learn how to use Spring Integrator?

[garyrussell]

@gkohen I started implementing it; as I hinted above, the challenge is managing the offsets of the replies.

Let's say I have 2 "in-flight" requests (and expect 2 replies).

The replies start coming in

• reply 1a topic1.partition0.offset0
• reply 2a topic1.partition0.offset1
• reply 1b topic1.partition0.offset2
  (2b is delayed for some reason).

So request 1 is complete, but we can't commit the offsets because the first part of request 2 would be committed while the request is still "in-flight". It gets more complicated when there are more topics/partitions involved.

However, thinking about it some more, I think we will be safe if we only commit offsets when there are no in-flight requests in process. If uncommitted offsets are subsequently replayed, they would just be discarded because there is no pending request in the super class.

[garyrussell]

PoC PR issued.