카프카 스프링 통합 흐름에 널 포인터가 있음

Question

카프카를 사용하여 스프링 통합 플로우를 구현하려고합니다. 그러나 나는 그 실수에 매달렸다.

디버깅 할 때 "HandlerAdapter handlerMethod"가 null 인 것으로 나타납니다. MessagingMessageListenerAdapter입니다. 내가 여기서하고 있는지, 아니면 더 많은 설정이 필요하다는 것을 확신하지 못한다. 그러나 이것은 문서와 다른 페이지에서 찾을 수있는 것입니다. 어떤 도움이

java.lang.NullPointerException: null 
at org.springframework.kafka.listener.adapter.MessagingMessageListenerAdapter.invokeHandler(MessagingMessageListenerAdapter.java:255) ~[spring-kafka-2.0.1.RELEASE.jar:2.0.1.RELEASE] 
at org.springframework.kafka.listener.adapter.RecordMessagingMessageListenerAdapter.onMessage(RecordMessagingMessageListenerAdapter.java:80) ~[spring-kafka-2.0.1.RELEASE.jar:2.0.1.RELEASE] 
at org.springframework.kafka.listener.adapter.RecordMessagingMessageListenerAdapter.onMessage(RecordMessagingMessageListenerAdapter.java:51) ~[spring-kafka-2.0.1.RELEASE.jar:2.0.1.RELEASE] 
at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.doInvokeRecordListener(KafkaMessageListenerContainer.java:923) [spring-kafka-2.0.1.RELEASE.jar:2.0.1.RELEASE] 
at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.doInvokeWithRecords(KafkaMessageListenerContainer.java:903) [spring-kafka-2.0.1.RELEASE.jar:2.0.1.RELEASE] 
at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeRecordListener(KafkaMessageListenerContainer.java:854) [spring-kafka-2.0.1.RELEASE.jar:2.0.1.RELEASE] 
at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeListener(KafkaMessageListenerContainer.java:729) [spring-kafka-2.0.1.RELEASE.jar:2.0.1.RELEASE] 
at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.run(KafkaMessageListenerContainer.java:615) [spring-kafka-2.0.1.RELEASE.jar:2.0.1.RELEASE] 
at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) [na:1.8.0_144] 
at java.util.concurrent.FutureTask.run(FutureTask.java:266) [na:1.8.0_144] 
at java.lang.Thread.run(Thread.java:748) [na:1.8.0_144] 

을 감사하는 것은 여기 내 pom.xml 파일

<?xml version="1.0" encoding="UTF-8"?> 
<project xmlns="http://maven.apache.org/POM/4.0.0" 
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" 
    xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 
http://maven.apache.org/xsd/maven-4.0.0.xsd"> 
<modelVersion>4.0.0</modelVersion> 

<groupId>com.cyclone.streams</groupId> 
<artifactId>kafka-poc</artifactId> 
<version>1.0</version> 

<parent> 
    <groupId>org.springframework.boot</groupId> 
    <artifactId>spring-boot-starter-parent</artifactId> 
    <version>1.5.8.RELEASE</version> 
</parent> 

<dependencies> 
    <!--Spring boot--> 
    <dependency> 
     <groupId>org.springframework.boot</groupId> 
     <artifactId>spring-boot-starter-web</artifactId> 
     <version>1.5.8.RELEASE</version> 
    </dependency> 
    <dependency> 
     <groupId>org.springframework.boot</groupId> 
     <artifactId>spring-boot-starter-integration</artifactId> 
     <version>1.5.8.RELEASE</version> 
    </dependency> 
    <dependency> 
     <groupId>org.springframework.boot</groupId> 
     <artifactId>spring-boot-starter-test</artifactId> 
     <scope>test</scope> 
    </dependency> 
    <!--Spring Kafka--> 
    <dependency> 
     <groupId>org.springframework.kafka</groupId> 
     <artifactId>spring-kafka</artifactId> 
     <version>2.0.1.RELEASE</version> 
    </dependency> 
    <dependency> 
     <groupId>org.springframework.integration</groupId> 
     <artifactId>spring-integration-kafka</artifactId> 
     <version>2.3.0.RELEASE</version> 
    </dependency> 
    <!--Other--> 
    <dependency> 
     <groupId>com.fasterxml.jackson.core</groupId> 
     <artifactId>jackson-databind</artifactId> 
    </dependency> 
</dependencies> 

<properties> 
    <java.version>1.8</java.version> 
</properties> 


<build> 
    <plugins> 
     <plugin> 
      <groupId>org.springframework.boot</groupId> 
      <artifactId>spring-boot-maven-plugin</artifactId> 
     </plugin> 
    </plugins> 
</build> 

<repositories> 
    <repository> 
     <id>spring-releases</id> 
     <url>https://repo.spring.io/libs-release</url> 
    </repository> 
</repositories> 
<pluginRepositories> 
    <pluginRepository> 
     <id>spring-releases</id> 
     <url>https://repo.spring.io/libs-release</url> 
    </pluginRepository> 
</pluginRepositories> 

그리고

@Configuration 
public class KafkaConsumerConfig { 

    @Bean 
    public IntegrationFlow flow() { 
     return IntegrationFlows 
       .from(kafkaMessageDrivenChannelAdapter()) 
       .get(); 
    } 

    @Bean 
    public KafkaMessageDrivenChannelAdapter<String, String> kafkaMessageDrivenChannelAdapter() { 
     KafkaMessageDrivenChannelAdapter<String, String> kafkaMessageDrivenChannelAdapter = 
       new KafkaMessageDrivenChannelAdapter<>(messageListenerContainer()); 
     kafkaMessageDrivenChannelAdapter.setOutputChannel(consumingChannel()); 
     return kafkaMessageDrivenChannelAdapter; 
    } 

    @Bean 
    public DirectChannel consumingChannel() { 
     return new DirectChannel(); 
    } 

    @Bean 
    public ConcurrentMessageListenerContainer<String, String> messageListenerContainer() { 
     ContainerProperties containerProps = new ContainerProperties("order-graph-timeline-topic"); 
     ConcurrentMessageListenerContainer<String, String> container = 
       new ConcurrentMessageListenerContainer(consumerFactory(), containerProps); 
     return container; 
    } 

    @Bean 
    ConcurrentKafkaListenerContainerFactory<Integer, String> kafkaListenerContainerFactory() { 
     ConcurrentKafkaListenerContainerFactory<Integer, String> factory = 
       new ConcurrentKafkaListenerContainerFactory<>(); 
     factory.setConsumerFactory(consumerFactory()); 
     return factory; 
    } 

    @Bean 
    public ConsumerFactory<Integer, String> consumerFactory() { 
     return new DefaultKafkaConsumerFactory<>(consumerConfigs()); 
    } 

    @Bean 
    public Map<String, Object> consumerConfigs() { 
     Map<String, Object> props = new HashMap<>(); 
     props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092"); 
     props.put(ConsumerConfig.GROUP_ID_CONFIG, "group");//TODO 
     props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false); 
     props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "100"); 
     props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "15000"); 
     props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, IntegerDeserializer.class); 
     props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class); 
     return props; 
    } 

} 

Answer 1

이 spring-integration-kafka-2.3.0.RELEASE처럼 보이는 코드입니다 spring-kafka-2.0.1.RELEASE와 호환되지 않습니다. 봄 부팅뿐만 아니라 1.5.8.RELEASE은 해당 버전과 호환되지 않습니다.

스프링 카프카 2.0을 사용해야 할 강력한 이유가 있습니까? 이것은 스프링 5.0베이스 라인을 위해 설계되었습니다.

스프링 카프카 1.3.1.RELEASE을 사용하는 것을 고려하십시오. Apache Kafka 0.11 +와 완벽하게 호환되지만, 동시에 언급 된 Spring Integration Kafka 2.3과 호환됩니다.