apache kafka 异常 - 获取 apikey: 3 和 apiVersion: 2 请求时出错

Question

我对 Apache Kafka 和 Spring Kafka 相当陌生，正在编写一个应用程序，它使用主题中的信息并执行一些操作。

我使用@KafkaListener 注释编写了一个简单的消费者，并针对我的本地 Kafka 服务器 0.10.1.1 对其进行了测试，它运行良好。

现在，当我指向我们的 Kafka DEV 服务器（版本 - 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(RequestChannel.scala:95)
        at kafka.network.RequestChannel$Request.<init>(RequestChannel.scala:87)
        at kafka.network.Processor$$anonfun$processCompletedReceives.apply(SocketServer.scala:488)
        at kafka.network.Processor$$anonfun$processCompletedReceives.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(RequestChannel.scala:92)

查了几个，基本是客户端和服务端的版本问题，没找到解决办法。

Answer 1

我通过在 pom.xml.

中覆盖下面的依赖项 kafka-clients 来让它工作

          <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka-clients</artifactId>
            <version>0.10.0.1</version>
            <scope>compile</scope>
            <exclusions>
                <exclusion>
                    <artifactId>slf4j-api</artifactId>
                    <groupId>org.slf4j</groupId>
                </exclusion>
            </exclusions>
        </dependency>

Answer 2

如果代理版本不匹配，也会发生此错误。我最近运行遇到了这个错误（在一个 3 节点集群中），其中 1 个代理在版本 1.0.0 上启动，另外 2 个在 0.10.x 上启动。将所有代理与版本 1.0.0 对齐解决了这个问题。

apache kafka 异常 - 获取 apikey: 3 和 apiVersion: 2 请求时出错

Exception on apache kafka - Error getting request for apikey: 3 and apiVersion: 2

apache-kafka

spring-kafka