Spring boot Kafka consumer

debugcn 投稿 Dev

suman j

Using spring integration Kafka dsl, I wonder why listener not receive messages? But the same application If I replace spring integration DSL with a method annotated with KafkaListener is able to consume messages fine. What am I missing with DSL?

DSL code that does not consume:

@Configuration
@EnableKafka
class KafkaConfig {
    //consumer factory provided by Spring boot
    @Bean
    IntegrationFlow inboundKafkaEventFlow(ConsumerFactory consumerFactory) {
        IntegrationFlows
                .from(Kafka
                .messageDrivenChannelAdapter(consumerFactory, "kafkaTopic")
                .configureListenerContainer({ c -> c.groupId('kafka-consumer-staging') })
                .id("kafkaTopicListener").autoStartup(true)
        )

                .channel("logChannel")
                .get()
    }
}

logChannel (or any other channel I use), does not reflect inbound messages.

Instead of the above code, If I use plain listener, it works fine to consume messages.

@Component
class KafkaConsumer {
    @KafkaListener(topics = ['kafkaTopic'], groupId = 'kafka-consumer-staging')
    void inboundKafkaEvent(String message) {
        log.debug("message is {}", message)
    }
}

Both approaches uses same application.properties for Kafka consumer.

Artem Bilan

You are missing the fact that you use Spring Integration, but you haven't enable it in your application. You don't need to do that for Kafka though, since you are not going to consume it with the @KafkaListener. So, to enable Spring Integration infrastructure, you need to add @EnableIntegration on your @Configuration class: https://docs.spring.io/spring-integration/docs/5.1.6.RELEASE/reference/html/#configuration-enable-integration

この記事はインターネットから収集されたものであり、転載の際にはソースを示してください。

侵害の場合は、連絡してください[email protected]