Spring Kafka无法找到Kafka属性Bean。

huangapple
go评论120阅读模式
英文:

Spring Kafka cannot find Kafka Properties Bean

问题

我正在尝试在我的项目中使用 Spring Kafka，其中 Spring Boot 版本为 2.2.2.RELEASE，Kafka 版本为 2.3.x。

  1. implementation 'org.springframework.kafka:spring-kafka'

我已经设置了 KafkaConfig 类和 Listener 类：

  1. @EnableKafka
  2. @Configuration
  3. public class KafkaConfig {
  4.     Config cfg = new Config();
  6.     @Bean
  7.     public ConsumerFactory<String, String> consumerFactory() {
  8.         Map<String, Object> props = new HashMap<>();
  9.         props.put(
  10.             ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
  11.             cfg.getProperty("server"));
  12.         props.put(
  13.             ConsumerConfig.GROUP_ID_CONFIG,
  14.             UUID.randomUUID());
  15.         props.put(
  16.             ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
  17.             StringDeserializer.class);
  18.         props.put(
  19.             ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
  20.             StringDeserializer.class);
  21.         return new DefaultKafkaConsumerFactory<>(props);
  22.     }
  24.     @Bean
  25.     public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() {
  26.         ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory();
  27.         factory.setConsumerFactory(consumerFactory());
  28.         return factory;
  29.     }
  30. }

KafkaConsumer.java

  1. @Service
  2. public class KafkaConsumer {
  3.     private final Logger LOG = LoggerFactory.getLogger(KafkaConsumer.class);
  5.     @KafkaListener(topics = "anomaly-topic")
  6.     void listener(String message) {
  7.         LOG.info("Listener [{}]", message);
  8.     }
  9. }

当我尝试运行我的应用程序时，我得到以下异常：

  1. org.springframework.beans.factory.UnsatisfiedDependencyException: Error creating bean with name 'org.springframework.boot.autoconfigure.kafka.KafkaAnnotationDrivenConfiguration': Unsatisfied dependency expressed through constructor parameter 0; nested exception is org.springframework.beans.factory.BeanCreationException: Error creating bean with name 'spring.kafka-org.springframework.boot.autoconfigure.kafka.KafkaProperties': Instantiation of bean failed; nested exception is org.springframework.beans.BeanInstantiationException: Failed to instantiate [org.springframework.boot.autoconfigure.kafka.KafkaProperties]: Constructor threw exception; nested exception is java.lang.NoClassDefFoundError: org/apache/kafka/common/requests/IsolationLevel
  2.     at org.springframework.beans.factory.support.ConstructorResolver.createArgumentArray(ConstructorResolver.java:798) ~[spring-beans-5.2.2.RELEASE.jar:5.2.2.RELEASE]
英文:

I am trying to use Spring Kafka in my project where is spring boot version is 2.2.2.RELEASE, kafka version is 2.3.x.

  1.     implementation &#39;org.springframework.kafka:spring-kafka&#39;

I have my KafkaConfig class and Listener Class setup:

  1. @EnableKafka
  2. @Configuration
  3. public class KafkaConfig {
  4. Config cfg = new Config();
  7. @Bean
  8. public ConsumerFactory&lt;String, String&gt; consumerFactory() {
  9. Map&lt;String, Object&gt; props = new HashMap&lt;&gt;();
  10. props.put(
  11.     ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
  12.     cfg.getProperty(&quot;server&quot;));
  13. props.put(
  14.     ConsumerConfig.GROUP_ID_CONFIG,
  15.     UUID.randomUUID());
  16. props.put(
  17.     ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
  18.     StringDeserializer.class);
  19. props.put(
  20.     ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
  21.     StringDeserializer.class);
  22. return new DefaultKafkaConsumerFactory&lt;&gt;(props);
  23. }
  25. @Bean
  26. public ConcurrentKafkaListenerContainerFactory&lt;String, String&gt; kafkaListenerContainerFactory() {
  27. ConcurrentKafkaListenerContainerFactory&lt;String, String&gt; factory = new ConcurrentKafkaListenerContainerFactory();
  28. factory.setConsumerFactory(consumerFactory());
  29. return factory;
  30. }
  31. }

KafkaConsumer.java

  1. @Service
  2. public class KafkaConsumer {
  3. private final Logger LOG = LoggerFactory
  4.   .getLogger(KafkaConsumer.class);
  6. @KafkaListener(topics = &quot;anomaly-topic&quot;)
  7. void listener(String message) {
  8. LOG.info(&quot;Listener [{}]&quot;, message);
  9. }
  10. }

When i am trying to run my application, i am getting below exception:

  1. org.springframework.beans.factory.UnsatisfiedDependencyException: Error creating bean with name &#39;org.springframework.boot.autoconfigure.kafka.KafkaAnnotationDrivenConfiguration&#39;: Unsatisfied dependency expressed through constructor parameter 0; nested exception is org.springframework.beans.factory.BeanCreationException: Error creating bean with name &#39;spring.kafka-org.springframework.boot.autoconfigure.kafka.KafkaProperties&#39;: Instantiation of bean failed; nested exception is org.springframework.beans.BeanInstantiationException: Failed to instantiate [org.springframework.boot.autoconfigure.kafka.KafkaProperties]: Constructor threw exception; nested exception is java.lang.NoClassDefFoundError: org/apache/kafka/common/requests/IsolationLevel
  2. at org.springframework.beans.factory.support.ConstructorResolver.createArgumentArray(ConstructorResolver.java:798) ~[spring-beans-5.2.2.RELEASE.jar:5.2.2.RELEASE]

答案1

得分: 4

我发现解决方法是禁用KafkaAutoConfiguration.class，

@SpringBootApplication(exclude = KafkaAutoConfiguration.class)

英文:

I found the solution to this is to disable KafkaAutoConfiguration.class,

  1. @SpringBootApplication(exclude = KafkaAutoConfiguration.class)

答案2

得分: 1

以下是翻译好的部分：

我也遇到了同样的情况，原因是因为我使用了以下库：

  1. implementation(group="io.confluent", name="kafka-avro-serializer", version="5.5.1")

该库自带了其自己的 kafka-clients，版本为 5.5.1-ccs，这被引入并导致了兼容性问题。

我通过使用Spring依赖管理来解决这个问题：

  1. plugins {
  2.     id("org.springframework.boot") version "2.2.2.RELEASE"
  3.     id("io.spring.dependency-management") version "1.0.9.RELEASE"
  4. }

然后运行 gradle clean

英文:

I had the same happen to me, the reason was because I had the following library:

  1. implementation(group=&quot;io.confluent&quot;, name=&quot;kafka-avro-serializer&quot;, version=&quot;5.5.1&quot;)

Which comes with its own kafka-clients with version 5.5.1-ccs which was being picked up and causing compatibility issues.

I solved this by using Spring Dependency Management:

  1. plugins {
  2.     id(&quot;org.springframework.boot&quot;) version &quot;2.2.2.RELEASE&quot;
  3.     id(&quot;io.spring.dependency-management&quot;) version &quot;1.0.9.RELEASE&quot;
  4. }

Then running a gradle clean

huangapple
  • 本文由 发表于 2020年10月22日 20:04:52
  • 转载请务必保留本文链接：https://go.coder-hub.com/64481838.html
  • apache-kafka
  • java
  • spring-boot
  • spring-kafka
匿名

发表评论

匿名网友

:?: :razz: :sad: :evil: :!: :smile: :oops: :grin: :eek: :shock: :???: :cool: :lol: :mad: :twisted: :roll: :wink: :idea: :arrow: :neutral: :cry: :mrgreen:

确定