【发布时间】:2018-10-11 22:57:56
【问题描述】:
我对 Apache Kafka 和 Spring Kafka 相当陌生,并且正在编写一个应用程序,它使用来自主题的信息并执行一些操作。
我使用@KafkaListener 注解编写了一个简单的消费者,并针对我的本地 Kafka 服务器 0.10.1.1 对其进行了测试,结果一切正常。
现在,当我指向我们的 Kafka 开发服务器(版本 - 0.10.0.2.5)时,我看到客户端日志中出现以下异常:
network.Selector - Connection with qm-alpha.activehealth.net/10.42.53.152 disconnected
java.io.EOFException: null
at org.apache.kafka.common.network.NetworkReceive.readFromReadableChannel(NetworkReceive.java:83) ~[kafka-clients-0.10.1.1.jar!/:?]
at org.apache.kafka.common.network.NetworkReceive.readFrom(NetworkReceive.java:71) ~[kafka-clients-0.10.1.1.jar!/:?]
at org.apache.kafka.common.network.KafkaChannel.receive(KafkaChannel.java:154) ~[kafka-clients-0.10.1.1.jar!/:?]
at org.apache.kafka.common.network.KafkaChannel.read(KafkaChannel.java:135) ~[kafka-clients-0.10.1.1.jar!/:?]
at org.apache.kafka.common.network.Selector.pollSelectionKeys(Selector.java:343) [kafka-clients-0.10.1.1.jar!/:?]
at org.apache.kafka.common.network.Selector.poll(Selector.java:291) [kafka-clients-0.10.1.1.jar!/:?]
at org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:260) [kafka-clients-0.10.1.1.jar!/:?]
at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:232) [kafka-clients-0.10.1.1.jar!/:?]
at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:180) [kafka-clients-0.10.1.1.jar!/:?]
at org.apache.kafka.clients.consumer.internals.AbstractCoordinator.ensureCoordinatorReady(AbstractCoordinator.java:193) [kafka-clients-0.10.1.1.jar!/:?]
at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.poll(ConsumerCoordinator.java:248) [kafka-clients-0.10.1.1.jar!/:?]
at org.apache.kafka.clients.consumer.KafkaConsumer.pollOnce(KafkaConsumer.java:1013) [kafka-clients-0.10.1.1.jar!/:?]
at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:979) [kafka-clients-0.10.1.1.jar!/:?]
at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.run(KafkaMessageListenerContainer.java:532) [spring-kafka-1.1.2.RELEASE.jar!/:?]
at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) [?:1.8.0_121]
at java.util.concurrent.FutureTask.run(FutureTask.java:266) [?:1.8.0_121]
at java.lang.Thread.run(Thread.java:745) [?:1.8.0_121]
另外,我在服务器端日志中看到以下错误:
kafka.network.InvalidRequestException: Error getting request for apiKey: 3 and apiVersion: 2
at kafka.network.RequestChannel$Request.liftedTree2$1(RequestChannel.scala:95)
at kafka.network.RequestChannel$Request.<init>(RequestChannel.scala:87)
at kafka.network.Processor$$anonfun$processCompletedReceives$1.apply(SocketServer.scala:488)
at kafka.network.Processor$$anonfun$processCompletedReceives$1.apply(SocketServer.scala:483)
at scala.collection.Iterator$class.foreach(Iterator.scala:727)
at scala.collection.AbstractIterator.foreach(Iterator.scala:1157)
at scala.collection.IterableLike$class.foreach(IterableLike.scala:72)
at scala.collection.AbstractIterable.foreach(Iterable.scala:54)
at kafka.network.Processor.processCompletedReceives(SocketServer.scala:483)
at kafka.network.Processor.run(SocketServer.scala:413)
at java.lang.Thread.run(Thread.java:745)
Caused by: java.lang.IllegalArgumentException: Invalid version for API key 3: 2
at org.apache.kafka.common.protocol.ProtoUtils.schemaFor(ProtoUtils.java:31)
at org.apache.kafka.common.protocol.ProtoUtils.requestSchema(ProtoUtils.java:44)
at org.apache.kafka.common.protocol.ProtoUtils.parseRequest(ProtoUtils.java:60)
at org.apache.kafka.common.requests.MetadataRequest.parse(MetadataRequest.java:96)
at org.apache.kafka.common.requests.AbstractRequest.getRequest(AbstractRequest.java:48)
at kafka.network.RequestChannel$Request.liftedTree2$1(RequestChannel.scala:92)
我查了几下,发现这基本上是因为客户端和服务器之间的版本问题,但找不到解决这个问题的方法。
【问题讨论】:
-
看起来 0.10.0.x 版本的 kafka 客户端与 0.10.1.x 代理并不真正兼容,发现 Spring kafka 版本 1.1.2.RELEASE 使用 kafka 客户端 0.10.1.1 这里是哪里我找到了信息:groups.google.com/forum/#!topic/confluent-platform/TyOMM8hDXOw
-
好的 - 我通过覆盖依赖项来让它工作,实际上是使用 kafka-clients 版本 0.10.0.1 重新声明依赖项。但是我想知道这是否是正确的方法,或者是否有更好的方法来做到这一点?
-
这是正确的方法。 Kafka 客户与经纪人非常紧密地联系在一起。我们将很快为 0.10.2.x 客户端发布 1.2 版本。 2.0 快照已经支持这一点。2.0 M1 即将推出。
-
仅供参考:自 Kafka 0.10.2 以来,Kafka 客户端不再与经纪人紧密联系。展望未来,Kafka 客户端将与 Kafka 代理向后兼容和向前兼容。示例:您可以将 0.10.2 客户端与 0.10.1 代理一起使用(这是新的!),一旦 0.10.3 或 0.11.0 发布,0.10.2 也将与这些较新的代理一起使用。见cwiki.apache.org/confluence/display/KAFKA/…