无法使用Spring Boot构建Kafka消费者。

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

Failed to construct kafka consumer with Spring Boot

问题

我已经实现了一个简单的消费者应用程序来消费主题中的消息。当我运行kafka-consumer应用程序时，出现了以下错误。

堆栈跟踪

  1. org.springframework.context.ApplicationContextException: Failed to start bean 'org.springframework.kafka.config.internalKafkaListenerEndpointRegistry'; nested exception is org.apache.kafka.common.KafkaException: Failed to construct kafka consumer
  2.     at org.springframework.context.support.DefaultLifecycleProcessor.doStart(DefaultLifecycleProcessor.java:185) ~[spring-context-5.2.9.RELEASE.jar:5.2.9.RELEASE]
  4. org.apache.kafka.common.KafkaException: class org.apache.kafka.common.serialization.StringSerializer is not an instance of org.apache.kafka.common.serialization.Deserializer

配置类

  1. @Configuration
  2. @EnableKafka
  3. public class KafkaConfig {
  5.     private ConsumerFactory<String,String> consumerFactory()
  6.     {
  7.         Map<String,Object> config=new ConcurrentHashMap<>();
  9.         config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"127.0.0.1:9092");
  10.         config.put(ConsumerConfig.GROUP_ID_CONFIG,"group_string");
  11.         config.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringSerializer.class);
  12.         config.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,StringSerializer.class);
  14.         return new DefaultKafkaConsumerFactory<>(config);
  15.     }
  17.     @Bean
  18.     public ConcurrentKafkaListenerContainerFactory<String,String> kafkaListenerContainerFactory()
  19.     {
  20.         ConcurrentKafkaListenerContainerFactory<String,String> factory
  21.                 =new ConcurrentKafkaListenerContainerFactory<>();
  22.         factory.setConsumerFactory(consumerFactory());
  24.         return factory;
  25.     }
  26. }

监听器类

  1. @Component
  2. public class KafkaConsumer {
  4.     @KafkaListener(topics = {"Kafka_Example"},groupId = "group_string")
  5.     public void consume(String message)
  6.     {
  7.         System.out.println("Consumed Message "+message);
  8.     }
  9. }

pom.xml

  1. <dependency>
  2.     <groupId>org.springframework.boot</groupId>
  3.     <artifactId>spring-boot-starter-web</artifactId>
  4. </dependency>
  6. <dependency>
  7.     <groupId>org.springframework.kafka</groupId>
  8.     <artifactId>spring-kafka</artifactId>
  9.     <version>2.6.0</version>
  10. </dependency>
  11. <dependency>
  12.     <groupId>org.springframework.kafka</groupId>
  13.     <artifactId>spring-kafka-test</artifactId>
  14.     <version>2.6.0</version>
  15.     <scope>test</scope>
  16. </dependency>
  18. <dependency>
  19.     <groupId>com.fasterxml.jackson.core</groupId>
  20.     <artifactId>jackson-databind</artifactId>
  21.     <version>2.11.1</version>
  22. </dependency>

注意 - 我的Kafka版本是2.13-2.6.0。

英文:

I have implemented a simple consumer application to consume messages from the topics. When I run the kafka-consumer application then the following error has occurred.

StackTrace

  1. org.springframework.context.ApplicationContextException: Failed to start bean &#39;org.springframework.kafka.config.internalKafkaListenerEndpointRegistry&#39;; nested exception is org.apache.kafka.common.KafkaException: Failed to construct kafka consumer
  2. 	at org.springframework.context.support.DefaultLifecycleProcessor.doStart(DefaultLifecycleProcessor.java:185) ~[spring-context-5.2.9.RELEASE.jar:5.2.9.RELEASE]
  4. org.apache.kafka.common.KafkaException: class org.apache.kafka.common.serialization.StringSerializer is not an instance of org.apache.kafka.common.serialization.Deserializer

Config class

  1. @Configuration
  2. @EnableKafka
  3. public class KafkaConfig {
  5.     private ConsumerFactory&lt;String,String&gt; consumerFactory()
  6.     {
  7.         Map&lt;String,Object&gt; config=new ConcurrentHashMap&lt;&gt;();
  9.         config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,&quot;127.0.0.1:9092&quot;);
  10.         config.put(ConsumerConfig.GROUP_ID_CONFIG,&quot;group_string&quot;);
  11.         config.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringSerializer.class);
  12.         config.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,StringSerializer.class);
  14.         return new DefaultKafkaConsumerFactory&lt;&gt;(config);
  15.     }
  17.     @Bean
  18.     public ConcurrentKafkaListenerContainerFactory&lt;String,String&gt; kafkaListenerContainerFactory()
  19.     {
  20.         ConcurrentKafkaListenerContainerFactory&lt;String,String&gt; factory
  21.                 =new ConcurrentKafkaListenerContainerFactory&lt;&gt;();
  22.         factory.setConsumerFactory(consumerFactory());
  24.         return factory;
  25.     }
  26. }

Listener class

  1. @Component
  2. public class KafkaConsumer {
  4.     @KafkaListener(topics = {&quot;Kafka_Example&quot;},groupId = &quot;group_string&quot;)
  5.     public void consume(String message)
  6.     {
  7.         System.out.println(&quot;Consumed Message &quot;+message);
  8.     }
  9. }

pom.xml

  1. &lt;dependency&gt;
  2.             &lt;groupId&gt;org.springframework.boot&lt;/groupId&gt;
  3.             &lt;artifactId&gt;spring-boot-starter-web&lt;/artifactId&gt;
  4.         &lt;/dependency&gt;
  6.         &lt;dependency&gt;
  7.             &lt;groupId&gt;org.springframework.kafka&lt;/groupId&gt;
  8.             &lt;artifactId&gt;spring-kafka&lt;/artifactId&gt;
  9.             &lt;version&gt;2.6.0&lt;/version&gt;
  10.         &lt;/dependency&gt;
  11.  &lt;dependency&gt;
  12.             &lt;groupId&gt;org.springframework.kafka&lt;/groupId&gt;
  13.             &lt;artifactId&gt;spring-kafka-test&lt;/artifactId&gt;
  14.             &lt;version&gt;2.6.0&lt;/version&gt;
  15.             &lt;scope&gt;test&lt;/scope&gt;
  16.         &lt;/dependency&gt;
  18.         &lt;dependency&gt;
  19.             &lt;groupId&gt;com.fasterxml.jackson.core&lt;/groupId&gt;
  20.             &lt;artifactId&gt;jackson-databind&lt;/artifactId&gt;
  21.             &lt;version&gt;2.11.1&lt;/version&gt;
  22.         &lt;/dependency&gt;

Note - My Kafka version is 2.13-2.6.0

答案1

得分: 4

你正在使用 StringSerializer，但应该使用 StringDeserializer，一个用于序列化，另一个用于反序列化。
而且，由于你将它们设置为 ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG 和 ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG，显然你想要进行反序列化。

  1. config.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
  2. config.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
英文:

You are using a StringSerializer but should use a StringDeserializer, one serializes, the other deserializes.
And since you set them for ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG and ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG you apparently want to deserialize.

  1. config.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
  2. config.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);

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

发表评论

匿名网友

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

确定