Can Spring Kafka @KafkaListener support multiple consume

[xzhan]

Background
we have two separate Kafka clusters with different endpoint, but same topic prefix
cluster1: kafka-01.com topic topic-kafka01
cluster2: kafka-02.com topic topic-kafka02

but from consumer side it only deploy one pool which must connect both clusters

Current Behavior
we should write two @KafkaListener with config two different kafka endpoin on DefaultKafkaConsumerFactory.

We Wanted
can one listener support multiple consume

@garyrussell @artembilan

I am not sure whether we can or not . or we say whether we have this kind of ability , if no have , or we can't implement it , any suggestion

appreciate you can reply and comments

[artembilan]

Since those Kafka clusters are fully independent it is really natural and logical that yo are going to have different and independent connectors to them. Those DefaultKafkaConsumerFactory and their ConcurrentKafkaListenerContainerFactory.
And of course this leads to different @KafkaListener methods because they are going to use those different container factories.

Those different (and similar) @KafkaListener methods may just delegate to the same business method avoiding the logic duplication directly in the @KafkaListener .

My point is: why to ask about surrogate configuration in spring-kafka for the scenario which you cannot address on Apache Kafka brokers?
Since you deal with different clusters, you don't have choice and deal with different @KafkaListener s.

Another way probably is to have several instances of your microservice, but connected to different clusters each.
This way you won't have a duplication logic at all and you probably won't need to override Spring Boot auto-configuration for different clusters.
Each microservice instance would get its own Kafka cluster connection, but the logic inside is just the same.

I'm closing this as Invalid: we don't support that and we won't since it does not fit to the whole solution on top of those clusters.

[garyrussell]

Note that @KafkaListener is a repeatable annotation so you can add 2 to the same method.

If the only difference in the configuration is the cluster (and maybe a few other changes), you can simply override the bootstrap.servers property on one of them:

@KafkaListener(...)
@KafkaListener(..., properties = "bootstrap.servers=someOtherHost1:9092,someOtherHost2:9092")
void someMethod(...) {
}

You can use property placeholders there too ${other.brokers}.

If there are a lot of different configurations then 2 factories would be easier.

[xzhan]

thx @artembilan if we can't do it on tech view point to support multi consume, we really can consider to do it on infra level which means we deploy pair micro service to consume different cluster. that's also my original idea

[xzhan]

Note that @KafkaListener is a repeatable annotation so you can add 2 to the same method.

If the only difference in the configuration is the cluster (and maybe a few other changes), you can simply override the bootstrap.servers property on one of them:

@KafkaListener(...)
@KafkaListener(..., properties = "bootstrap.servers=someOtherHost1:9092,someOtherHost2:9092")
void someMethod(...) {
}

You can use property placeholders there too ${other.brokers}.

If there are a lot of different configurations then 2 factories would be easier.

thx @garyrussell , we are connecting target kafka cluster deploy on AWS with mTLS. that's property would be complex as we need to add keystore and trustore.

but thinking the difference we can use on keystore and truststore to connect different kafka cluster with config on factory.

so problem here would be : one factory , how to support different cluster connect or we need to define two factory which only bootstrap.servers config not same other is same, then name different factory and config on @KafkaListener

@KafkaListener(... containerFactory = "factory1")
@KafkaListener(..., containerFactory = "factory2")
void someMethod(...) {
}

So I am thinking can containerFactory natural support connect different cluster if it have different cluster server address?

[garyrussell]

You can externalize many properties; you don't need to specify them individually:

@KafkaListener(id = "kgh2114", topics = "kgh2114", properties = "${props}")
void listen(String in) {
	System.out.println(in);
}

props=bootstrap.servers=localhost:9092\n \
 max.poll.records=20\n \
 max.poll.interval.ms=120000

[xzhan]

You can externalize many properties; you don't need to specify them individually:

@KafkaListener(id = "kgh2114", topics = "kgh2114", properties = "${props}")
void listen(String in) {
	System.out.println(in);
}

props=bootstrap.servers=localhost:9092\n \
 max.poll.records=20\n \
 max.poll.interval.ms=120000

thx @garyrussell , if I understanding correct, we will do like this ?

@KafkaListener(id = "kgh2114", topics = "kgh2114", properties = "${props2114}")
 @KafkaListener(id = "kgh2115", topics = "kgh2114", properties = "${props2115}")
void listen(String in) {
	System.out.println(in);
}

 max.poll.records=20\n \
 max.poll.interval.ms=120000`

 max.poll.records=20\n \
 max.poll.interval.ms=120000`

[garyrussell]

Yes, with properties; if you are using application.yml, the syntax is

props: |-
  bootstrap.servers=localhost:9092
  max.poll.records=20
  max.poll.interval.ms=120000