Klasse nicht org.apache.kafka.clients.consumer.RangeAssignor im Frühjahr XD-Modul mit Spring Kafka Integration mit XML-Konfiguration

Question

Ich erhalte die Folge Ausnahme gefunden, wenn ich einen Kafka Verbraucher in meinem xd Modul versuchen zu konstruieren:

Caused by: org.apache.kafka.common.KafkaException: Failed to construct kafka consumer 
at org.apache.kafka.clients.consumer.KafkaConsumer.<init>(KafkaConsumer.java:702) ~[na:na] 
at org.apache.kafka.clients.consumer.KafkaConsumer.<init>(KafkaConsumer.java:557) ~[na:na] 
at org.springframework.kafka.core.DefaultKafkaConsumerFactory.createKafkaConsumer(DefaultKafkaConsumerFactory.java:73) ~[na:na] 
at org.springframework.kafka.core.DefaultKafkaConsumerFactory.createConsumer(DefaultKafkaConsumerFactory.java:69) ~[na:na] 
at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.<init>(KafkaMessageListenerContainer.java:284) ~[na:na] 
at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.<init>(KafkaMessageListenerContainer.java:222) ~[na:na] 
at org.springframework.kafka.listener.KafkaMessageListenerContainer.doStart(KafkaMessageListenerContainer.java:179) ~[na:na] 
at org.springframework.kafka.listener.AbstractMessageListenerContainer.start(AbstractMessageListenerContainer.java:204) ~[na:na] 
at org.springframework.context.support.DefaultLifecycleProcessor.doStart(DefaultLifecycleProcessor.java:173) ~[spring-context-4.1.7.RELEASE.jar:4.1.7.RELEASE] 
... 27 common frames omitted 
Caused by: org.apache.kafka.common.KafkaException: org.apache.kafka.clients.consumer.RangeAssignor ClassNotFoundException exception occured 
at org.apache.kafka.common.config.AbstractConfig.getConfiguredInstances(AbstractConfig.java:227) ~[na:na] 
at org.apache.kafka.clients.consumer.KafkaConsumer.<init>(KafkaConsumer.java:637) ~[na:na] 
... 35 common frames omitted 
Caused by: java.lang.ClassNotFoundException: org.apache.kafka.clients.consumer.RangeAssignor 
at java.net.URLClassLoader.findClass(URLClassLoader.java:381) ~[na:1.8.0_77] 
at java.lang.ClassLoader.loadClass(ClassLoader.java:424) ~[na:1.8.0_77] 
at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:331) ~[na:1.8.0_77] 
at java.lang.ClassLoader.loadClass(ClassLoader.java:357) ~[na:1.8.0_77] 
at java.lang.Class.forName0(Native Method) ~[na:1.8.0_77] 
at java.lang.Class.forName(Class.java:348) ~[na:1.8.0_77] 
at org.apache.kafka.common.utils.Utils.newInstance(Utils.java:332) ~[na:na] 
at org.apache.kafka.common.config.AbstractConfig.getConfiguredInstances(AbstractConfig.java:225) 

Hier ist meine xml:

<int-kafka:message-driven-channel-adapter 
    id="kafkaListener" listener-container="customKafkaMessageListenerContainer" 
    auto-startup="false" phase="100" send-timeout="5000" channel="toTransformer" 
    error-channel="errorChannel"/> 

<bean id="containerProps" 
    class="org.springframework.kafka.listener.config.ContainerProperties"> 
    <constructor-arg name="topicPartitions" ref="topicPartitionInitialOffset" /> 
</bean> 

<bean id="topicPartitionInitialOffset" 
    class="org.springframework.kafka.support.TopicPartitionInitialOffset"> 
    <constructor-arg index="0" name="topic" value="source-8-26" 
     type="java.lang.String" /> 
    <constructor-arg index="1" name="partition" value="0" 
     type="int" /> 
    <constructor-arg index="2" name="initialOffset" value="0" 
     type="java.lang.Long" /> 
</bean> 

<bean id="consumerFactory" 
    class="org.springframework.kafka.core.DefaultKafkaConsumerFactory"> 
    <constructor-arg> 
     <map> 
      <entry key="bootstrap.servers" value="localhost:9092" /> 
      <entry key="group.id" value="test-consumer-group" /> 
      <entry key="autocommit.enable" value="false" /> 
      <entry key="value.deserializer" value-type="java.lang.Class" value="org.apache.kafka.common.serialization.StringSerializer"> 
      </entry> 
      <entry key="key.deserializer" value-type="java.lang.Class" value="org.apache.kafka.common.serialization.StringSerializer"> 
      </entry> 
     </map> 
    </constructor-arg> 
</bean> 

<bean id="customKafkaMessageListenerContainer" 
    class="org.springframework.kafka.listener.KafkaMessageListenerContainer"> 
    <constructor-arg name="consumerFactory" ref="consumerFactory"> 
    </constructor-arg> 
    <constructor-arg name="containerProperties" ref="containerProps" /> 
</bean> 

Abhängigkeiten:

<dependencies> 
    <dependency> 
     <groupId>org.springframework.integration</groupId> 
     <artifactId>spring-integration-kafka</artifactId> 
     <version>2.0.1.RELEASE</version> 
    </dependency> 
    <dependency> 
     <groupId>org.apache.kafka</groupId> 
     <artifactId>kafka-clients</artifactId> 
     <version>0.10.0.1</version> 
    </dependency> 
    <dependency> 
     <groupId>org.apache.kafka</groupId> 
     <artifactId>kafka_2.11</artifactId> 
     <version>0.10.0.1</version> 
    </dependency> 
    <dependency> 
     <groupId>org.springframework.kafka</groupId> 
     <artifactId>spring-kafka</artifactId> 
     <version>1.0.3.RELEASE</version> 
    </dependency> 
</dependencies> 

weiß jemand, warum Ich bekomme diese Klasse nicht gefunden Ausnahme? Ich überprüfte die API für Kafka 0.10.0 und die Klasse existiert, also bin ich nicht sicher, warum ich diese Ausnahme empfange. Könnte es damit zu tun haben, wie Spring XD die Klassen lädt? Ich weiß, dass Spring XD mit Kafka 0.8.2 in seinem lib-Ordner geliefert wird, also lädt Spring XD vielleicht zuerst Kafka 0.8.2 und kann dann die Klasse nicht finden?

Jeder Einblick würde sehr geschätzt werden!

Answer 1

Ich würde erwarten, dass es funktioniert, ok, weil das Modul in einem eigenen Klassenlader geladen ist, der die richtigen Klassen aus dem lib-Ordner des Moduls finden sollte.

Ich debuggen normalerweise diese Art von Problem, indem Sie die JVM mit -verbose ausführen, die das Laden aller Klassen protokolliert.

Hinweis: Wenn Sie Kafka für den XD-Transport verwenden, unterstützt Spring XD nur den Client 0.8.x.x - die Folgeprojekte (Spring Cloud Stream/Spring Cloud Dataflow) unterstützen die neueren kafka-Clients. Die first milestone of version 1.1 supporting the new clients was announced last week.