获取 InvalidRecordException

Question

在我的应用程序中，我获取使用 kafka 排队并保存在磁盘上的数据，以及从 kafka 获取此数据并进行处理的消费者。但是当我的消费者试图从 kafka 读取数据时，我遇到了以下异常：

2017-06-09 10:57:24,733 ERROR NetworkClient Uncaught error in request completion:
org.apache.kafka.common.KafkaException: Error deserializing key/value for partition TcpMessage-1 at offset 155884487
at org.apache.kafka.clients.consumer.internals.Fetcher.parseRecord(Fetcher.java:628) ~[kafka-clients-0.9.0.1.jar:?]
at org.apache.kafka.clients.consumer.internals.Fetcher.handleFetchResponse(Fetcher.java:566) ~[kafka-clients-0.9.0.1.jar:?]
at org.apache.kafka.clients.consumer.internals.Fetcher.access[=10=]0(Fetcher.java:69) ~[kafka-clients-0.9.0.1.jar:?]
at org.apache.kafka.clients.consumer.internals.Fetcher.onSuccess(Fetcher.java:139) ~[kafka-clients-0.9.0.1.jar:?]
at org.apache.kafka.clients.consumer.internals.Fetcher.onSuccess(Fetcher.java:136) ~[kafka-clients-0.9.0.1.jar:?]
at org.apache.kafka.clients.consumer.internals.RequestFuture.fireSuccess(RequestFuture.java:133) ~[kafka-clients-0.9.0.1.jar:?]
at org.apache.kafka.clients.consumer.internals.RequestFuture.complete(RequestFuture.java:107) ~[kafka-clients-0.9.0.1.jar:?]
at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient$RequestFutureCompletionHandler.onComplete(ConsumerNetworkClient.java:380) ~[kafka-clients-0.9.0.1.jar:?]
at org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:274) [kafka-clients-0.9.0.1.jar:?]
at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.clientPoll(ConsumerNetworkClient.java:320) [kafka-clients-0.9.0.1.jar:?]
at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:213) [kafka-clients-0.9.0.1.jar:?]
at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:193) [kafka-clients-0.9.0.1.jar:?]
at org.apache.kafka.clients.consumer.KafkaConsumer.pollOnce(KafkaConsumer.java:908) [kafka-clients-0.9.0.1.jar:?]
at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:853) [kafka-clients-0.9.0.1.jar:?]
at com.affirmed.mediation.edr.kafka.tcpMessage.TcpMessageConsumer.doWork(TcpMessageConsumer.java:190) [EdrServer.jar:?]
at com.affirmed.mediation.edr.kafka.tcpMessage.TcpMessageConsumer.run(TcpMessageConsumer.java:248) [EdrServer.jar:?]

Caused by: org.apache.kafka.common.record.InvalidRecordException: Record is corrupt (stored crc = 2016852547, computed crc = 1399853379)
at org.apache.kafka.common.record.Record.ensureValid(Record.java:226) ~[kafka-clients-0.9.0.1.jar:?]***
at org.apache.kafka.clients.consumer.internals.Fetcher.parseRecord(Fetcher.java:617) ~[kafka-clients-0.9.0.1.jar:?]
... 15 more

谁能帮我解决这个问题。我被它困住了，无法找出根源。

Answer 1

I got stuck with it and not able to figure out the root.

堆栈跟踪已经告诉您确切的问题是什么：

2017-06-09 10:57:24,733 ERROR NetworkClient Uncaught error in request completion:
org.apache.kafka.common.KafkaException: Error deserializing key/value for partition TcpMessage-1 at offset 155884487
...
Caused by: org.apache.kafka.common.record.InvalidRecordException: Record is corrupt (stored crc = 2016852547, computed crc = 1399853379)

根本问题是输入主题 TcpMessage 的分区 1 中消息偏移量 155884487 的数据已损坏。这表明服务于该主题的分区 1 的 Kafka 代理的本地存储存在问题，或者存在网络问题（代理的本地存储正常，但是当消费者通过网络读取数据时，数据在运输途中损坏）。

获取 InvalidRecordException

Getting InvalidRecordException

apache-kafka

kafka-consumer-api

kafka-producer-api

apache-kafka-connect