Can Spring-Kafka do smart routing base on message to different Kafka clusters

[xzhan]

Problem define:
We have different Kafka clusters , one is building on AWS , one is on our internal DC.
here is our call flow diagram, Wx internal DC consume service can only connect internal DC Kafka , we want to build one routing service base on some static rule to forward message from AWS Cluster kafka to Internal DC Kafka

Example
AWS cluster Kafka send message bring Header (Key :domain, Value: domain001 )
Router service parse Header base on domain001 query DC mapping then send to target Internal DC Kafka

@garyrussell do you have any idea on this kind of case ? can we base on record header to do routing ? can spring-kafka have well support on this case?

[xzhan]

Hi @garyrussell can you take look it

[artembilan]

This one has the same story and same reporter: #2022.

As I said over here: Spring Integration is a library you need for your solution.
The routing logic is really out of Spring for Apache Kafka project scope.

[garyrussell]

There is nothing built-in to support this; if you don't want to use Spring Integration for some reason, it would be simple for you to create a DelegatingKafkaTemplate to decide which template to use to publish a message based on some criteria.

[xzhan]

thx @garyrussell and @artembilan , I am try to contact with confluent team about how to use Kafka Stream processor API to handle it , I think Spring-Kafka and Spring-Integration can handle well also

[xzhan]

Hmm Kafka-Stream it is only for one cluster , not allow write to another DC , I am looking back to this problem

[xzhan]

@garyrussell can you show me what's concept for DelegatingKafkaTemplate

[garyrussell]

public class MyTemplate {

	private final Map<String, KafkaTemplate<String, String>> templates;

	public MyTemplate(Map<String, KafkaTemplate<String, String>> templates) {
		this.templates = new HashMap<>(templates);
	}

	public ListenableFuture<SendResult<String, String>> send(String topic, String key,
			@Nullable String data) {

		KafkaTemplate<String, String> kafkaTemplate = templates.get(topic);
		Assert.notNull(kafkaTemplate, "No template for topic " + topic);
		return kafkaTemplate.send(topic, key, data);
	}

}

[xzhan]

thx @garyrussell our criteria for select KafkaTemplate it is base on record header value.
Header:
objectType: [OrgID|SiteUUID]
id: 4cde993b-bbd7-4ffd-a25d-392ebdbb743a

then our different business listener consumer different Kafka cluster build on AWS with mTLS enable.

let me describe it more clearly with call flow

follow you idea our DelegatingKafkaTemplate have much more logical.
Base on header to find routing DC info
Case 1: objectType is OrgID query from remote Service get the mapping of OrgID's Kafka DC then cache on local
Case2: objectType is SiteUUID query from remote Service get SiteUUID's Kafka DC, then cache on local
after that we use suitable target KafkaTemplate to send message

How can we design well on the DelegateKafkaTemplate? can you show me some idea and suggestion for easy Expansion and maintenance @garyrussell

[garyrussell]

Just do as I suggested, but use one of the other send() methods (ProducerRecord or Message<?>).

[xzhan]

Yes @garyrussell follow you idea. we need to construct one smartMessageConvert to change ConsumerRecord<K,V> to ProducerRecord<K,V> also bring header info (routing info on header). after than our @KafakListener receive message then can base on DelegatingTemplate to send message with interface send(ProducerRecord). Do i understand well for it ? any sample code to show how MessageConvert work and Intergrating with ConcurrentKafkaListenerContainerFactory

[garyrussell]

You can't do the conversion on the consumer side; the infrastructure needs consumer records; there is code in the DeadLetterPublishingRecoverer that converts a failed Consumer Record to a ProducerRecord to send it to a dead letter topic, if you need an example.

[xzhan]

@garyrussell Yes I also noticed that function on DeadLetterPublishingRecoverer. At same time as you mentioned above we use send() with Message<?> object can do it . so I am trying with this on consumer Side

@KafkaListener(..)

public void listen(List<Message<?>> listMessages, Acknowledgment ack) {
   
 kafkaTemplate.send(listMessage); like this to directly use  Message<?> object to send

}

I am testing on this , it may save us effort to convert ConsumerRecord to ProducerRecord. right ? but I found I have no chance to change the topic if i want to send to another Topic

->>>>>>>>>>>>>>>>>>>>>>>>>>>>><<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<

but for another interface
@KafkaListener(..)

public void listen(List<ConsumerRecord<?, ?>> records, Acknowledgment ack) {

try {
records.forEach(consumerRecord -> {
sendProduceRecord(consumerRecord);
});
} catch (Exception e) {

    } finally {
        ack.acknowledge();
    }
}

}

here is my change from consumerRecord to producerRecord: which I can change the topic if i want ,also the record.value we change to our target Message.class. it works well
protected ProducerRecord<Object, Object> createProducerRecord(ConsumerRecord record) {
try {
BaseMessage retryMessage = messageParser.parse(record.value().toString(), RetryMessage.class);
return new ProducerRecord(webexSJDCTopic,
(record.partition() < 0) ? null : Integer.valueOf(record.partition()),
record.key(), retryMessage, record.headers());
} catch (KafkaMessageParseException e) {
logger.error("KafkaMessageParseException exception " ,e);
}
logger.error("ConvertMessage failure exception ");
return null;
}

[xzhan]

@garyrussell , Hi I have successfully build DelegatingKafkaTemplate which implement AppliationContextAware , it will preload all related kafkaTemplated in to one Map ,

@Component
public class DelegatingKafkaTemplate implements ApplicationContextAware {
 private ApplicationContext applicationContext;
    private Map<String, ***KafkaTemplate> ***KafkaTemplateMap;


    @PostConstruct
    void  load***KafkaTemplateMap() {
        ***KafkaTemplateMap = applicationContext.getBeansOfType(***KafkaTemplate.class);
    }


...

how about you think the design ? thx

[garyrussell]

You shouldn't use getBeansOfType() in a @PostConstruct method - it's too early and might cause early instantiation of other beans. If you must do it there, use the version with two booleans getBeansOfType(WebexKafkaTemplate.class, false, false); but it would be better to implement SmartLifecycle and do it in start().

[xzhan]

Educated @garyrussell . I will change it and follow it

[xzhan]

@garyrussell I report one interest issue to you #2067

[xzhan]

Hi @garyrussell I found we can do it like this
@component
public class DelegatingKafkaTemplate {

@Autowired
private Map<String, ***KafkaTemplate> ***KafkaTemplateMap;

}

spring will auto inject related beans into this map, how about this ?