将字节流反序列化为 Scala 案例类对象时出现 java.lang.Instantiation 异常答案

【问题标题】：java.lang.Instantiation Exception while deserializing a byte stream into a Scala case class object将字节流反序列化为 Scala 案例类对象时出现 java.lang.Instantiation 异常
【发布时间】：2019-10-15 14:43:53
【问题描述】：

我正在尝试将 avro 字节流反序列化为 scala 案例类对象。基本上，我有一个带有 avro 编码数据流动的 kafka 流，现在模式中有一个补充，我正在尝试更新 scala 案例类以包含新字段。案例类是这样的

/** Case class to hold the Device data. */
case class DeviceData(deviceId: String,
                sw_version: String,
                timestamp: String,
                reading: Double,
                new_field: Option[String] = None
               )  {

this() = this("na", "na", "na", 0, 无) }

avro架构如下：

{
  "type": "record",
  "name": "some_name",
  "namespace": "some_namespace",
  "fields": [
    {
      "name": "deviceId",
      "type": "string"
    },
    {
      "name": "sw_version",
      "type": "string"
    }, 
    {
      "name": "timestamp",
      "type": "string"
    },
    {
      "name": "reading",
      "type": "double"
    },
    {
      "name": "new_field",
     "type": ["null", "string"],
      "default": null
    }]}

收到数据后出现以下异常：

java.lang.RuntimeException: java.lang.InstantiationException

我可以很好地接收用 python 编写的消费者的数据，所以我知道数据正在以正确的格式正确流式传输。我怀疑问题出在案例类构造函数的创建上，我尝试过这样做：

/** Case class to hold the Device data. */
case class DeviceData(deviceId: String,
                sw_version: String,
                timestamp: String,
                reading: Double,
                new_field: Option[String]
               )  {
this() = this("na", "na", "na", 0, some("na"))
}

但没有运气。

反序列化代码是（摘录）：

// reader and decoder for reading avro records
private var reader: DatumReader[T] = null
private var decoder : BinaryDecoder = null
decoder = DecoderFactory.get.binaryDecoder(message, decoder)
reader.read(null.asInstanceOf[T], decoder)

我找不到用于反序列化 avro 的案例类的构造函数的任何其他示例，我去年发布了一个相关问题 java.lang.NoSuchMethodException for init method in Scala case class 并且根据响应我能够实现我当前的代码从那以后工作正常。

【问题讨论】：

avro 架构在哪里？你有一个null 解码器......一个null 阅读器......那么decoder 和reader 在这个问题中确实很重要。除非您的 decoder 和 reader 定义正确，否则它将无法正常工作。
我已经添加了 avro 架构，请参见上文。空读取器和解码器也被正确定义，这是所有工作代码，除了“new_field”的添加导致java异常，我尝试添加一个 println(decoder.getString()) 并且我能够看到第一个字符串字段（只是为了查看异常是由 DecoderFactory 还是读取器部分抛出的，它是由 reader.read 抛出的）。

标签： scala apache-kafka apache-flink avro

【解决方案1】：

我采用完全不同的方法解决了这个问题。我使用了本示例 https://github.com/jfrazee/schema-registry-examples/tree/master/src/main/scala/io/atomicfinch/examples/flink 中提供的 Confluent Kafka 客户端。我还有一个 Confluent 模式注册表，使用 kafka 和模式注册表 https://docs.confluent.io/current/quickstart/ce-docker-quickstart.html 附带的容器化多合一解决方案非常容易设置。

我必须在我的 pom.xml 文件中添加融合依赖项和 repo。这在存储库部分。

<repository>
    <id>confluent</id>
    <url>http://packages.confluent.io/maven/</url>
</repository>

这在依赖部分：

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-avro-confluent-registry</artifactId>
    <version>1.8.0</version>
</dependency>
<dependency>
    <groupId>io.confluent</groupId>
    <artifactId>kafka-avro-serializer</artifactId>
    <!-- For Confluent Platform 5.2.1 -->
    <version>5.2.1</version>
</dependency>

使用https://github.com/jfrazee/schema-registry-examples/blob/master/src/main/scala/io/atomicfinch/examples/flink/ConfluentRegistryDeserializationSchema.scala 中提供的代码，我能够与 Confluent 模式注册表对话，然后根据 avro 消息头中的模式 id 从模式 reg 下载模式并返回一个 GenericRecord 对象，我从中可以轻松地任何和所有感兴趣的领域，并创建一个新的 DeviceData 对象的 DataStream。

val kafka_consumer = new FlinkKafkaConsumer010("prod.perfwarden.minute",
  new ConfluentRegistryDeserializationSchema[GenericRecord](classOf[GenericRecord], "http://localhost:8081"),
  properties)
val device_data_stream = env
  .addSource(kafka_consumer)
  .map({x => new DeviceData(x.get("deviceId").toString,
    x.get("sw_version").toString,
    x.get("timestamp").toString,
    x.get("reading").toString.toDouble,
    x.get("new_field").toString)})

confluent kafka 客户端负责根据架构反序列化 avro 字节流，包括默认值。设置模式注册表和使用 confluent kafka 客户端可能需要一点时间来适应，但可能是更好的长期解决方案，只需我的 2 美分。

【讨论】：