How Can One Microservice Consume Multiple Kafka Topics

Axon Framework
,
1

PROBLEM: We have an Axon Command Microservice that has to listen to four Kafka topics. But there is only one parameter where we define a topic: axon.kafka.default-topic=my.topic

RESEARCH: We have SpringBoot and Axon autoconfiguration enabled. axon.kafka.default-topic works fine if one microservice listens to another.

QUESTION: How to listen to several Kafka topics?

2

Hi @zonin

I don’t know if you have tried, but to fully support Kafka since you are using Springboot what about use spring-cloud-stream?

I have being using on my projects it allows me to even connect on two kafka clusters not only multiple topics

3

Resulted in an overriden Bean:

@Bean
public StreamableKafkaMessageSource<String, byte[]> streamableKafkaMessageSource(
  ConsumerFactory<String, byte[]> kafkaConsumerFactory,
  Fetcher<String, byte[], KafkaEventMessage> kafkaFetcher) {

return StreamableKafkaMessageSource.<String, byte[]>builder()
    .topics(Arrays.asList(topics))
    .consumerFactory(kafkaConsumerFactory)
    .fetcher(kafkaFetcher)
    .build();

}

Any improvements and optimizations are welcome.

4

What you’ve done would actually be the most reasonable solution for the time being.
Essentially there’s a desire to introduce a feature which allows you to map events to certain topics, based on the message converter present in the extension.

As the extension is still on a milestone release though, we haven’t come around to providing new functionality. Added, I tend to hear not a lot of noise on it anymore, meaning the prioritization of it decreases somewhat…

On another note, curious to hear how you’re using the Kafka Extension @zonin. Have you set it up to actually connect distinct Axon clients, or as a means to tie into the company wide Kafka topic?

5

Kafka topics are for Command-side microservices to post their events. A topic per Command Microservice. Other Command Microservices and Query Microservices subscribe these topics when required.

Non-AxonServer solution unfortunately, because customer needs PostgreSQL and Kafka. So Aggregate State, Kafka events stored as blobs in DB by Axon, along with other Axon data.

6

Gotcha @zonin, thanks for sharing that. Sometimes client requirements impose constraints you’d like to work around of course… Still valuable information though, also for others I assume.

Hoping a future client will allow you to play with Axon Server too :wink: