将带有参数的案例类作为案例类转换为 Avro 消息以发送到 Kafka

Question

我正在使用具有嵌套案例类和Seq[Nested Case Classes] 的案例类 问题是当我尝试使用KafkaAvroSerializer 对其进行序列化时，它会抛出：

Caused by: java.lang.IllegalArgumentException: Unsupported Avro type. Supported types are null, Boolean, Integer, Long, Float, Double, String, byte[] and IndexedRecord
    at io.confluent.kafka.serializers.AbstractKafkaAvroSerDe.getSchema(AbstractKafkaAvroSerDe.java:115)
    at io.confluent.kafka.serializers.AbstractKafkaAvroSerializer.serializeImpl(AbstractKafkaAvroSerializer.java:71)
    at io.confluent.kafka.serializers.KafkaAvroSerializer.serialize(KafkaAvroSerializer.java:54)
    at org.apache.kafka.common.serialization.Serializer.serialize(Serializer.java:60)
    at org.apache.kafka.clients.producer.KafkaProducer.doSend(KafkaProducer.java:879)
    at org.apache.kafka.clients.producer.KafkaProducer.send(KafkaProducer.java:841)
    at org.apache.kafka.clients.producer.KafkaProducer.send(KafkaProducer.java:728)```

Answer 1

如果您想将 Avro 与案例类等 Scala 构造一起使用，我建议您使用 Avro4s。这对所有 scala 特性都有本机支持，如果你想要的话，甚至可以从你的模型中创建模式。

有一些自动类型类派生的陷阱。这是我学到的。

至少使用 avro4s 版本 2.0.4

一些宏会生成带有编译器警告的代码，并且还会破坏疣去除器。我们必须添加以下注释来编译我们的代码（有时错误是找不到隐含的，但它是由宏生成代码中的错误引起的）：

@com.github.ghik.silencer.silent
@SuppressWarnings(Array("org.wartremover.warts.Null", "org.wartremover.warts.AsInstanceOf", "org.wartremover.warts.StringPlusAny"))

下一个自动类型类派生一次只能工作一个级别。我创建了一个对象来保存我的架构的所有SchemaFor、Decoder 和Encoder 实例。然后我从最内部的类型开始明确地构建类型类实例。我还使用implicitly 来验证每个 ADT 将在移动到下一个之前解决。例如：

sealed trait Notification
object Notification {
  final case class Outstanding(attempts: Int) extends Notification
  final case class Complete(attemts: Int, completedAt: Instant) extends Notification
}

sealed trait Job
final case class EnqueuedJob(id: String, enqueuedAt: Instant) extends Job
final case class RunningJob(id: String, enqueuedAt: Instant, startedAt: Instant) extends Job
final case class FinishedJob(id: String, enqueuedAt: Instant, startedAt: Instant, completedAt: Instant) extends Job

object Schema {

  // Explicitly define schema for ADT instances
  implicit val schemaForNotificationComplete: SchemaFor[Notification.Complete] = SchemaFor.applyMacro
  implicit val schemaForNotificationOutstanding: SchemaFor[Notification.Outstanding] = SchemaFor.applyMacro

  // Verify Notification ADT is defined
  implicitly[SchemaFor[Notification]]
  implicitly[Decoder[Notification]]
  implicitly[Encoder[Notification]]

  // Explicitly define schema, decoder and encoder for ADT instances
  implicit val schemaForEnqueuedJob: SchemaFor[EnqueuedJob] = SchemaFor.applyMacro
  implicit val decodeEnqueuedJob: Decoder[EnqueuedJob] = Decoder.applyMacro
  implicit val encodeEnqueuedJob: Encoder[EnqueuedJob] = Encoder.applyMacro

  implicit val schemaForRunningJob: SchemaFor[RunningJob] = SchemaFor.applyMacro
  implicit val decodeRunningJob: Decoder[RunningJob] = Decoder.applyMacro
  implicit val encodeRunningJob: Encoder[RunningJob] = Encoder.applyMacro

  implicit val schemaForFinishedJob: SchemaFor[FinishedJob] = SchemaFor.applyMacro
  implicit val decodeFinishedJob: Decoder[FinishedJob] = Decoder.applyMacro
  implicit val encodeFinishedJob: Encoder[FinishedJob] = Encoder.applyMacro

  // Verify Notification ADT is defined
  implicitly[Encoder[Job]]
  implicitly[Decoder[Job]]
  implicitly[SchemaFor[Job]]

  // And so on until complete nested ADT is defined
}