Kafka Avro Deserializer 无法将 Kafka 消息反序列化为特定的 Avro 记录答案

【问题标题】：Kafka Avro Deserializer is not able to deserialize the Kafka message to Specific Avro RecordKafka Avro Deserializer 无法将 Kafka 消息反序列化为特定的 Avro 记录
【发布时间】：2020-12-13 14:11:45
【问题描述】：

我正在尝试将 kafka 中的 Avro messgaes 反序列化为从 Avro Schema 生成的 POJO。我正在使用 KafkaAvroDeserializer 进行此转换。

我可以在 kafka 返回的ConsumerRecord<String, Data> 记录中看到GenericRecord。但是当我尝试将此记录分配给生成的 POJO 类对象时，date 类型的 POJO 字段与ClassCastException 失败。当我检查 avro 有效负载时，此日期字段以 integer 的形式出现。

设置： Avro - 1.9.1 融合 - 5.4 commercehub gradle 插件 0.20.0

在尝试反序列化 Avro 消息时，我收到错误消息 -

Caused by: org.apache.kafka.common.errors.SerializationException: Error deserializing Avro message for id 66
Caused by: java.lang.ClassCastException: java.lang.Integer cannot be cast to java.time.LocalDate
    at com.sample.Data.put(Data.java:229) ~[main/:na]
    at org.apache.avro.generic.GenericData.setField(GenericData.java:795) ~[avro-1.9.1.jar:1.9.1]
    at org.apache.avro.specific.SpecificDatumReader.readField(SpecificDatumReader.java:139) ~[avro-1.9.1.jar:1.9.1]
    at org.apache.avro.generic.GenericDatumReader.readRecord(GenericDatumReader.java:237) ~[avro-1.9.1.jar:1.9.1]
    at org.apache.avro.specific.SpecificDatumReader.readRecord(SpecificDatumReader.java:123) ~[avro-1.9.1.jar:1.9.1]
    at org.apache.avro.generic.GenericDatumReader.readWithoutConversion(GenericDatumReader.java:170) ~[avro-1.9.1.jar:1.9.1]
    at org.apache.avro.specific.SpecificDatumReader.readField(SpecificDatumReader.java:136) ~[avro-1.9.1.jar:1.9.1]
    at org.apache.avro.generic.GenericDatumReader.readRecord(GenericDatumReader.java:237) ~[avro-1.9.1.jar:1.9.1]
    at org.apache.avro.specific.SpecificDatumReader.readRecord(SpecificDatumReader.java:123) ~[avro-1.9.1.jar:1.9.1]
    at org.apache.avro.generic.GenericDatumReader.readWithoutConversion(GenericDatumReader.java:170) ~[avro-1.9.1.jar:1.9.1]
    at org.apache.avro.generic.GenericDatumReader.read(GenericDatumReader.java:151) ~[avro-1.9.1.jar:1.9.1]
    at org.apache.avro.generic.GenericDatumReader.read(GenericDatumReader.java:144) ~[avro-1.9.1.jar:1.9.1]
    at io.confluent.kafka.serializers.AbstractKafkaAvroDeserializer$DeserializationContext.read(AbstractKafkaAvroDeserializer.java:287) ~[kafka-avro-serializer-5.4.0.jar:na]
    at io.confluent.kafka.serializers.AbstractKafkaAvroDeserializer.deserialize(AbstractKafkaAvroDeserializer.java:102) ~[kafka-avro-serializer-5.4.0.jar:na]
    at io.confluent.kafka.serializers.AbstractKafkaAvroDeserializer.deserialize(AbstractKafkaAvroDeserializer.java:81) ~[kafka-avro-serializer-5.4.0.jar:na]
    at io.confluent.kafka.serializers.KafkaAvroDeserializer.deserialize(KafkaAvroDeserializer.java:55) ~[kafka-avro-serializer-5.4.0.jar:na]
    at org.apache.kafka.common.serialization.Deserializer.deserialize(Deserializer.java:60) ~[kafka-clients-2.5.0.jar:na]
    at org.apache.kafka.clients.consumer.internals.Fetcher.parseRecord(Fetcher.java:1310) ~[kafka-clients-2.5.0.jar:na]
    at org.apache.kafka.clients.consumer.internals.Fetcher.access$3500(Fetcher.java:128) ~[kafka-clients-2.5.0.jar:na]
    at org.apache.kafka.clients.consumer.internals.Fetcher$CompletedFetch.fetchRecords(Fetcher.java:1541) ~[kafka-clients-2.5.0.jar:na]
    at org.apache.kafka.clients.consumer.internals.Fetcher$CompletedFetch.access$1700(Fetcher.java:1377) ~[kafka-clients-2.5.0.jar:na]
    at org.apache.kafka.clients.consumer.internals.Fetcher.fetchRecords(Fetcher.java:677) ~[kafka-clients-2.5.0.jar:na]
    at org.apache.kafka.clients.consumer.internals.Fetcher.fetchedRecords(Fetcher.java:632) ~[kafka-clients-2.5.0.jar:na]
    at org.apache.kafka.clients.consumer.KafkaConsumer.pollForFetches(KafkaConsumer.java:1290) ~[kafka-clients-2.5.0.jar:na]
    at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1248) ~[kafka-clients-2.5.0.jar:na]
    at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1216) ~[kafka-clients-2.5.0.jar:na]
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.doPoll(KafkaMessageListenerContainer.java:1091) [spring-kafka-2.5.2.RELEASE.jar:2.5.2.RELEASE]
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.pollAndInvoke(KafkaMessageListenerContainer.java:1047) [spring-kafka-2.5.2.RELEASE.jar:2.5.2.RELEASE]
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.run(KafkaMessageListenerContainer.java:972) [spring-kafka-2.5.2.RELEASE.jar:2.5.2.RELEASE]
    at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) [na:1.8.0_241]
    at java.util.concurrent.FutureTask.run(FutureTask.java:266) [na:1.8.0_241]
    at java.lang.Thread.run(Thread.java:748) [na:1.8.0_241]

转换失败的字段的 Avro 架构 ClassCastException

{
    "name": "BIRTH_DT",
    "type": [
        "null",
        {
            "type": "int",
            "logicalType": "date"
        }
    ],
    "default": null
}

来自生成的 POJO 的代码片段

  @Deprecated public java.time.LocalDate BIRTH_DT;

  // Used by DatumReader.  Applications should not call.
  @SuppressWarnings(value="unchecked")
  public void put(int field$, java.lang.Object value$) {
    switch (field$) {
    .
    .
    case 8: BIRTH_DT = (java.time.LocalDate)value$; break;
    default: throw new org.apache.avro.AvroRuntimeException("Bad index");
    }
  }

  public java.time.LocalDate getBIRTHDT() {
    return BIRTH_DT;
  }

  public void setBIRTHDT(java.time.LocalDate value) {
      this.BIRTH_DT = value;
  }

Kafka 消费者方法

    @KafkaListener(topics = "${spring.kafka.consumer.properties.topic}",
                     groupId = "${spring.kafka.consumer.group-id}")
    // Data is a POJO generated by Avro tools
    public void consume(ConsumerRecord<String, Data> record,
                        @Header(KafkaHeaders.RECEIVED_PARTITION_ID) Integer partition,
                        @Header(KafkaHeaders.OFFSET) Long offset, Acknowledgment ack) throws IOException {
    
        logger.info(String.format("#### -> Consumed message -> partiion: %s, offset: %s", partition, offset));
        Data row = record.value();
        ack.acknowledge();
    }

build.gradle

buildscript {
    repositories {
        jcenter {
            url "https://nexus.abc.com:8443/content/repositories/jcenter/"
        }
    }
    dependencies {
        classpath "com.commercehub.gradle.plugin:gradle-avro-plugin:0.20.0"
    }
}

plugins {
    id 'org.springframework.boot' version '2.3.1.RELEASE'
    id 'io.spring.dependency-management' version '1.0.9.RELEASE'
    id 'java'
    id 'idea'
    id 'eclipse'
}

repositories {
    maven { url nexusPublicRepoURL }
    maven { url "https://nexus.abc.com:8443/content/repositories/confluence.io-maven/" }
    jcenter()
    maven { url "https://nexus.abc.com:8443/content/repositories/jcenter/" }
}

group = 'com.abc.cscm'
version = '0.0.1-SNAPSHOT'
sourceCompatibility = '8'
targetCompatibility = '8'

ext {
    springCloudVersion = 'Hoxton.SR6'
    confluentVersion = '5.4.0'
}

dependencies {
    implementation 'org.springframework.boot:spring-boot-starter-actuator'
    implementation 'org.springframework.boot:spring-boot-starter-web'
    implementation 'org.springframework.kafka:spring-kafka'

    implementation "io.confluent:kafka-avro-serializer:${confluentVersion}"

    implementation 'org.apache.avro:avro:1.9.1'

    testImplementation('org.springframework.boot:spring-boot-starter-test') {
        exclude group: 'org.junit.vintage', module: 'junit-vintage-engine'
    }
    testImplementation 'org.springframework.kafka:spring-kafka-test'
}

springBoot {
    buildInfo()
}

dependencyManagement {
    imports {
        mavenBom "org.springframework.cloud:spring-cloud-dependencies:${springCloudVersion}"
    }
}

test {
    useJUnitPlatform()
}

wrapper {
    distributionUrl = "https://nexus.abc.com:8443/service/local/repositories/thirdparty/content/org/gradle/gradle/6.5/gradle-6.5.zip"
}

apply plugin: "com.commercehub.gradle.plugin.avro"
apply plugin: 'idea'

./gradlew dependencies --configuration compileClasspath (Output)

> Task :dependencies

------------------------------------------------------------
Root project
------------------------------------------------------------

compileClasspath - Compile classpath for source set 'main'.
                        ** omiting spring deps
+--- io.confluent:kafka-avro-serializer:5.4.0
|    +--- org.apache.avro:avro:1.9.1
|    |    +--- com.fasterxml.jackson.core:jackson-core:2.9.9 -> 2.11.0
|    |    +--- com.fasterxml.jackson.core:jackson-databind:2.9.9.3 -> 2.11.0 (*)
|    |    +--- org.apache.commons:commons-compress:1.19
|    |    \--- org.slf4j:slf4j-api:1.7.25 -> 1.7.30
|    +--- io.confluent:kafka-schema-registry-client:5.4.0
|    |    +--- org.apache.kafka:kafka-clients:5.4.0-ccs -> 2.5.0 (*)
|    |    +--- io.confluent:common-config:5.4.0
|    |    |    +--- io.confluent:common-utils:5.4.0
|    |    |    |    \--- org.slf4j:slf4j-api:1.7.26 -> 1.7.30
|    |    |    \--- org.slf4j:slf4j-api:1.7.26 -> 1.7.30
|    |    +--- org.apache.avro:avro:1.9.1 (*)
|    |    +--- com.fasterxml.jackson.core:jackson-databind:2.9.10.1 -> 2.11.0 (*)
|    |    +--- io.swagger:swagger-annotations:1.5.22
|    |    +--- io.swagger:swagger-core:1.5.3
|    |    |    +--- org.apache.commons:commons-lang3:3.2.1 -> 3.10
|    |    |    +--- org.slf4j:slf4j-api:1.6.3 -> 1.7.30
|    |    |    +--- com.fasterxml.jackson.core:jackson-annotations:2.4.5 -> 2.11.0
|    |    |    +--- com.fasterxml.jackson.core:jackson-databind:2.4.5 -> 2.11.0 (*)
|    |    |    +--- com.fasterxml.jackson.datatype:jackson-datatype-joda:2.4.5 -> 2.11.0
|    |    |    |    +--- com.fasterxml.jackson.core:jackson-core:2.11.0
|    |    |    |    \--- joda-time:joda-time:2.9.9
|    |    |    +--- com.fasterxml.jackson.dataformat:jackson-dataformat-yaml:2.4.5 -> 2.11.0
|    |    |    |    +--- com.fasterxml.jackson.core:jackson-databind:2.11.0 (*)
|    |    |    |    +--- org.yaml:snakeyaml:1.26
|    |    |    |    \--- com.fasterxml.jackson.core:jackson-core:2.11.0
|    |    |    +--- io.swagger:swagger-models:1.5.3
|    |    |    |    +--- com.fasterxml.jackson.core:jackson-annotations:2.4.5 -> 2.11.0
|    |    |    |    +--- org.slf4j:slf4j-api:1.6.3 -> 1.7.30
|    |    |    |    \--- io.swagger:swagger-annotations:1.5.3 -> 1.5.22
|    |    |    \--- com.google.guava:guava:18.0 -> 29.0-android
|    |    |         +--- com.google.guava:failureaccess:1.0.1
|    |    |         +--- com.google.guava:listenablefuture:9999.0-empty-to-avoid-conflict-with-guava
|    |    |         +--- com.google.code.findbugs:jsr305:3.0.2
|    |    |         +--- org.checkerframework:checker-compat-qual:2.5.5
|    |    |         +--- com.google.errorprone:error_prone_annotations:2.3.4
|    |    |         \--- com.google.j2objc:j2objc-annotations:1.3
|    |    \--- io.confluent:common-utils:5.4.0 (*)
|    +--- io.confluent:common-config:5.4.0 (*)
|    \--- io.confluent:common-utils:5.4.0 (*)
\--- org.apache.avro:avro:1.9.1 (*)

./gradlew buildEnvironment（输出）

classpath
+--- com.commercehub.gradle.plugin:gradle-avro-plugin:0.20.0
|    \--- org.apache.avro:avro-compiler:1.9.2    <<<<<<<<<<<<<<<<<<<<<<<<<<
|         +--- org.apache.avro:avro:1.9.2
|         |    +--- com.fasterxml.jackson.core:jackson-core:2.10.2 -> 2.11.0
|         |    +--- com.fasterxml.jackson.core:jackson-databind:2.10.2 -> 2.11.0
|         |    |    +--- com.fasterxml.jackson.core:jackson-annotations:2.11.0
|         |    |    \--- com.fasterxml.jackson.core:jackson-core:2.11.0
|         |    +--- org.apache.commons:commons-compress:1.19
|         |    \--- org.slf4j:slf4j-api:1.7.25 -> 1.7.30
|         +--- org.apache.commons:commons-lang3:3.9 -> 3.10
|         +--- org.apache.velocity:velocity-engine-core:2.2
|         |    +--- org.apache.commons:commons-lang3:3.9 -> 3.10
|         |    \--- org.slf4j:slf4j-api:1.7.30
|         +--- com.fasterxml.jackson.core:jackson-databind:2.10.2 -> 2.11.0 (*)
|         +--- joda-time:joda-time:2.10.1
|         \--- org.slf4j:slf4j-api:1.7.25 -> 1.7.30

我不确定是否应该编辑生成的 POJO 类，或者我遗漏了什么。

我能够通过更改下面问题中提到的架构将 avro 消息转换为 POJO。但我认为这很hacky，问题还没有解决。

问题 - Avro is not able to deserialize Union with Logical Types in fields

【问题讨论】：

确保您的序列化器和反序列化器使用相同的日期类型对象
是的，序列化和反序列化使用相同的日期类型对象。

标签： java spring-boot apache-kafka avro confluent-platform

【解决方案1】：

您能否澄清一下您使用哪个版本的avro-maven-plugin 来生成 POJO？从 avro 版本 1.9.0 开始，Joda-Time 已被弃用，取而代之的是 Java8 JSR310，Java8 被设置为默认值。见Apache Avro 1.9.0 release notes。

当我从头开始生成 POJO 时，我得到 java.time.LocalDate BIRTH_DT 而不是 org.joda.time.LocalDate BIRTH_DT。

   @Deprecated public java.time.LocalDate BIRTH_DT;

因此，在您的情况下，我认为类路径中很可能存在 avro 版本不匹配或过时的 pojo。我建议通过 mvn dependency:tree -Dincludes=org.apache.avro:avro 调用验证 avro 版本并重新生成 POJO。

【讨论】：

您能提供您的 gradle 构建文件吗？谢谢。
感谢您的帮助。我正在使用支持 avro 1.9.0-1.9.2 的 com.commercehub.gradle.plugin:gradle-avro-plugin:0.20.0。现在，我使用了 gradle plugin ， avro dependency 来匹配 confluent kafka avro deserializer 5.4.0。当我执行 ./gradlew 依赖项时 --configuration compileClasspath。所有 avro 依赖项现在都是 1.9.1。正如你所说，我已经重新生成了 POJO，现在 BIRTH_DT 列显示为 java.time.LocalDate。运行错误后，我仍然收到这种新类型的类转换异常。
好的，知道了，谢谢。您能否也让我知道，记录在发送到 kafka 之前是如何序列化的？是否也是用于记录序列化的 avro 1.9.0 库？
我已按照您的建议使用新的 avro 设置更新了问题详细信息。还添加了 gradle 文件的一些输出。我还没有调查记录是如何写入 kafka 的，我认为这应该没关系，因为我们有模式。但是让我检查一下。
用于序列化 avro 1.8 正在使用。